From 7ee40bd485d131a03a5f85ed40ed182bdf6cf2fd Mon Sep 17 00:00:00 2001 From: WangchuXiao Date: Thu, 20 Jul 2023 16:28:58 +0800 Subject: [PATCH] fix: init grpc conn multiple in one process will cause zero address when rpc is called (#604) * new feature: add batch send msg * new feature: add batch send msg * new feature: add batch send msg * new feature: add batch send msg * new feature: add batch send msg * new feature: add batch send msg * fix bug: multiple gateway kick user * fix bug: multiple gateway kick user * fix bug: multiple gateway kick user * fix bug: multiple gateway kick user * fix bug: multiple gateway kick user * MsgDestructTime * fix bug: msg destruct sql * fix bug: msg destruct * fix bug: msg destruct * fix bug: msg destruct sql * fix bug: msg destruct sql * fix bug: msg destruct sql * fix bug: msg destruct sql * debug: print stack * debug: print stack * debug: print stack * fix bug: msg destruct sql Signed-off-by: wangchuxiao * fix bug: msg notification self 2 self push twice * fix bug: heartbeat get self notification * fix bug: init grpc conn in one process * fix bug: grpc conn Signed-off-by: wangchuxiao * fix bug: zk client recreate node when reconn * fix bug: set friend mark args error --------- Signed-off-by: wangchuxiao --- pkg/discoveryregistry/zookeeper/discover.go | 12 +++++++++++- pkg/discoveryregistry/zookeeper/register.go | 20 +++++++++++++++----- pkg/discoveryregistry/zookeeper/resolver.go | 9 +++++---- pkg/discoveryregistry/zookeeper/zk.go | 6 +++++- pkg/proto/friend/friend.go | 3 --- 5 files changed, 36 insertions(+), 14 deletions(-) diff --git a/pkg/discoveryregistry/zookeeper/discover.go b/pkg/discoveryregistry/zookeeper/discover.go index b56999911..f83331943 100644 --- a/pkg/discoveryregistry/zookeeper/discover.go +++ b/pkg/discoveryregistry/zookeeper/discover.go @@ -40,7 +40,17 @@ func (s *ZkClient) watch() { event := <-s.eventChan switch event.Type { case zk.EventSession: - s.logger.Printf("zk session event: %+v", event) + if event.State == zk.StateHasSession && s.isRegistered { + s.logger.Printf("zk session event stateHasSession: %+v, client prepare to create new temp node", event) + node, err := s.CreateTempNode(s.rpcRegisterName, s.rpcRegisterAddr) + if err != nil { + s.logger.Printf("zk session event stateHasSession: %+v, create temp node error: %v", event, err) + } else { + s.node = node + } + } else { + s.logger.Printf("zk session event: %+v", event) + } case zk.EventNodeChildrenChanged: s.logger.Printf("zk event: %s", event.Path) l := strings.Split(event.Path, "/") diff --git a/pkg/discoveryregistry/zookeeper/register.go b/pkg/discoveryregistry/zookeeper/register.go index fec72538d..8fc5fa34e 100644 --- a/pkg/discoveryregistry/zookeeper/register.go +++ b/pkg/discoveryregistry/zookeeper/register.go @@ -30,6 +30,14 @@ func (s *ZkClient) CreateRpcRootNodes(serviceNames []string) error { return nil } +func (s *ZkClient) CreateTempNode(rpcRegisterName, addr string) (node string, err error) { + return s.conn.CreateProtectedEphemeralSequential( + s.getPath(rpcRegisterName)+"/"+addr+"_", + []byte(addr), + zk.WorldACL(zk.PermAll), + ) +} + func (s *ZkClient) Register(rpcRegisterName, host string, port int, opts ...grpc.DialOption) error { if err := s.ensureName(rpcRegisterName); err != nil { return err @@ -39,15 +47,14 @@ func (s *ZkClient) Register(rpcRegisterName, host string, port int, opts ...grpc if err != nil { return err } - node, err := s.conn.CreateProtectedEphemeralSequential( - s.getPath(rpcRegisterName)+"/"+addr+"_", - []byte(addr), - zk.WorldACL(zk.PermAll), - ) + node, err := s.CreateTempNode(rpcRegisterName, addr) if err != nil { return err } + s.rpcRegisterName = rpcRegisterName + s.rpcRegisterAddr = addr s.node = node + s.isRegistered = true return nil } @@ -60,6 +67,9 @@ func (s *ZkClient) UnRegister() error { } time.Sleep(time.Second) s.node = "" + s.rpcRegisterName = "" + s.rpcRegisterAddr = "" + s.isRegistered = false s.localConns = make(map[string][]grpc.ClientConnInterface) s.resolvers = make(map[string]*Resolver) return nil diff --git a/pkg/discoveryregistry/zookeeper/resolver.go b/pkg/discoveryregistry/zookeeper/resolver.go index e395a8f34..7e9c8c125 100644 --- a/pkg/discoveryregistry/zookeeper/resolver.go +++ b/pkg/discoveryregistry/zookeeper/resolver.go @@ -73,7 +73,11 @@ func (s *ZkClient) Build( opts resolver.BuildOptions, ) (resolver.Resolver, error) { s.logger.Printf("build resolver: %+v, cc: %+v", target, cc.UpdateState) - // log.ZDebug(context.Background(), "build resolver start", "target", target, "cc", cc.UpdateState) + serviceName := strings.TrimLeft(target.URL.Path, "/") + if oldResolver, ok := s.resolvers[serviceName]; ok { + s.logger.Printf("rpc resolver exist: %+v, cc: %+v, key: %s", target, cc.UpdateState, serviceName) + return oldResolver, nil + } r := &Resolver{} r.target = target r.cc = cc @@ -81,11 +85,8 @@ func (s *ZkClient) Build( r.ResolveNowZK(resolver.ResolveNowOptions{}) s.lock.Lock() defer s.lock.Unlock() - serviceName := strings.TrimLeft(target.URL.Path, "/") s.resolvers[serviceName] = r s.logger.Printf("build resolver finished: %+v, cc: %+v, key: %s", target, cc.UpdateState, serviceName) - // log.ZDebug(context.Background(), "build resolver finished", "target", target, "cc", cc.UpdateState, - // "serviceName", serviceName) return r, nil } diff --git a/pkg/discoveryregistry/zookeeper/zk.go b/pkg/discoveryregistry/zookeeper/zk.go index a3b02b12e..cc333c4b6 100644 --- a/pkg/discoveryregistry/zookeeper/zk.go +++ b/pkg/discoveryregistry/zookeeper/zk.go @@ -40,7 +40,10 @@ type ZkClient struct { userName string password string - scheme string + rpcRegisterName string + rpcRegisterAddr string + isRegistered bool + scheme string timeout int conn *zk.Conn @@ -136,6 +139,7 @@ func NewClient(zkServers []string, zkRoot string, options ...ZkOption) (*ZkClien resolver.Register(client) go client.refresh() go client.watch() + time.Sleep(time.Millisecond * 50) return client, nil } diff --git a/pkg/proto/friend/friend.go b/pkg/proto/friend/friend.go index f2ad6a7a4..197fc753c 100644 --- a/pkg/proto/friend/friend.go +++ b/pkg/proto/friend/friend.go @@ -152,9 +152,6 @@ func (x *SetFriendRemarkReq) Check() error { if x.FriendUserID == "" { return errs.ErrArgs.Wrap("fromUserID is empty") } - if x.Remark == "" { - return errs.ErrArgs.Wrap("remark is empty") - } return nil }