fix:#1143 Feature/reduce etcd registry conn; wait group modify

.github/workflows/github-actions.yml

@@ -59,7 +59,7 @@ jobs:
         # diff -u <(echo -n) <(gofmt -d -s .)
 
     - name: Install go ci lint
-      run: curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(go env GOPATH)/bin v1.27.0
+      run: curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(go env GOPATH)/bin v1.41.1
 
     - name: Run Linter
       run: golangci-lint run --timeout=10m -v

registry/base_registry.go

@@ -122,9 +122,9 @@ func (r *BaseRegistry) Destroy() {
 	// first step close registry's all listeners
 	r.facadeBasedRegistry.CloseListener()
 	// then close r.done to notify other program who listen to it
-	close(r.done)
+	close(r.Done())
 	// wait waitgroup done (wait listeners outside close over)
-	r.wg.Wait()
+	r.WaitGroup().Wait()
 
 	// close registry client
 	r.closeRegisters()
@@ -474,7 +474,7 @@ func (r *BaseRegistry) closeRegisters() {
 // IsAvailable judge to is registry not closed by chan r.done
 func (r *BaseRegistry) IsAvailable() bool {
 	select {
-	case <-r.done:
+	case <-r.Done():
 		return false
 	default:
 		return true

registry/etcdv3/registry.go

@@ -95,10 +95,8 @@ func newETCDV3Registry(url *common.URL) (registry.Registry, error) {
 	); err != nil {
 		return nil, err
 	}
-	r.WaitGroup().Add(1) // etcdv3 client start successful, then wg +1
-
-	go etcdv3.HandleClientRestart(r)
 
+	r.handleClientRestart()
 	r.InitListeners()
 
 	return r, nil
@@ -175,3 +173,8 @@ func (r *etcdV3Registry) DoSubscribe(svc *common.URL) (registry.Listener, error)
 func (r *etcdV3Registry) DoUnsubscribe(conf *common.URL) (registry.Listener, error) {
 	return nil, perrors.New("DoUnsubscribe is not support in etcdV3Registry")
 }
+
+func (r *etcdV3Registry) handleClientRestart() {
+	r.WaitGroup().Add(1)
+	go etcdv3.HandleClientRestart(r)
+}

registry/kubernetes/registry.go

@@ -162,7 +162,6 @@ func newKubernetesRegistry(url *common.URL) (registry.Registry, error) {
 		return nil, perrors.WithStack(err)
 	}
 
-	r.WaitGroup().Add(1)
 	go r.HandleClientRestart()
 	r.InitListeners()
 
@@ -191,12 +190,12 @@ func newMockKubernetesRegistry(
 
 // HandleClientRestart will reconnect to  kubernetes registry center
 func (r *kubernetesRegistry) HandleClientRestart() {
+	r.WaitGroup().Add(1)
+	defer r.WaitGroup().Done()
 	var (
 		err       error
 		failTimes int
 	)
-
-	defer r.WaitGroup().Done()
 LOOP:
 	for {
 		select {

registry/zookeeper/registry.go

@@ -74,7 +74,7 @@ func newZkRegistry(url *common.URL) (registry.Registry, error) {
 		return nil, err
 	}
 
-	r.WaitGroup().Add(1) //zk client start successful, then wg +1
+	r.WaitGroup().Add(1)
 	go zookeeper.HandleClientRestart(r)
 
 	r.listener = zookeeper.NewZkEventListener(r.client)
@@ -108,7 +108,6 @@ func newMockZkRegistry(url *common.URL, opts ...gxzookeeper.Option) (*zk.TestClu
 	if err != nil {
 		return nil, nil, err
 	}
-	r.WaitGroup().Add(1) // zk client start successful, then wg +1
 	go zookeeper.HandleClientRestart(r)
 	r.InitListeners()
 	return c, r, nil
@@ -314,3 +313,8 @@ func (r *zkRegistry) getCloseListener(conf *common.URL) (*RegistryConfigurationL
 
 	return zkListener, nil
 }
+
+func (r *zkRegistry) handleClientRestart() {
+	r.WaitGroup().Add(1)
+	go zookeeper.HandleClientRestart(r)
+}

remoting/etcdv3/facade.go

@@ -18,19 +18,16 @@
 package etcdv3
 
 import (
-	getty "github.com/apache/dubbo-getty"
 	"sync"
 	"time"
 )
 
 import (
 	gxetcd "github.com/dubbogo/gost/database/kv/etcd/v3"
-	perrors "github.com/pkg/errors"
 )
 
 import (
 	"dubbo.apache.org/dubbo-go/v3/common"
-	"dubbo.apache.org/dubbo-go/v3/common/constant"
 	"dubbo.apache.org/dubbo-go/v3/common/logger"
 )
 
@@ -45,56 +42,18 @@ type clientFacade interface {
 }
 
 // HandleClientRestart keeps the connection between client and server
+// This method should be used only once. You can use handleClientRestart() in package registry.
 func HandleClientRestart(r clientFacade) {
-	var (
-		err       error
-		failTimes int
-	)
-
 	defer r.WaitGroup().Done()
-LOOP:
 	for {
 		select {
+		case <-r.Client().GetCtx().Done():
+			r.RestartCallBack()
+			// re-register all services
+			time.Sleep(10 * time.Microsecond)
 		case <-r.Done():
 			logger.Warnf("(ETCDV3ProviderRegistry)reconnectETCDV3 goroutine exit now...")
-			break LOOP
-			// re-register all services
-		case <-r.Client().Done():
-			r.ClientLock().Lock()
-			clientName := gxetcd.RegistryETCDV3Client
-			timeout, _ := time.ParseDuration(r.GetURL().GetParam(constant.REGISTRY_TIMEOUT_KEY, constant.DEFAULT_REG_TIMEOUT))
-			endpoints := r.Client().GetEndPoints()
-			r.Client().Close()
-			r.SetClient(nil)
-			r.ClientLock().Unlock()
-
-			// try to connect to etcd,
-			failTimes = 0
-			for {
-				after := getty.GetTimeWheel().After(timeSecondDuration(failTimes * gxetcd.ConnDelay))
-				select {
-				case <-r.Done():
-					logger.Warnf("(ETCDV3ProviderRegistry)reconnectETCDRegistry goroutine exit now...")
-					break LOOP
-				case <-after: // avoid connect frequent
-				}
-				err = ValidateClient(
-					r,
-					gxetcd.WithName(clientName),
-					gxetcd.WithEndpoints(endpoints...),
-					gxetcd.WithTimeout(timeout),
-					gxetcd.WithHeartbeat(1),
-				)
-				logger.Infof("ETCDV3ProviderRegistry.validateETCDV3Client(etcd Addr{%s}) = error{%#v}",
-					endpoints, perrors.WithStack(err))
-				if err == nil && r.RestartCallBack() {
-					break
-				}
-				failTimes++
-				if gxetcd.MaxFailTimes <= failTimes {
-					failTimes = gxetcd.MaxFailTimes
-				}
-			}
+			return
 		}
 	}
 }

remoting/zookeeper/facade.go

@@ -42,6 +42,7 @@ type ZkClientFacade interface {
 }
 
 // HandleClientRestart keeps the connection between client and server
+// This method should be used only once. You can use handleClientRestart() in package registry.
 func HandleClientRestart(r ZkClientFacade) {
 	defer r.WaitGroup().Done()
 	for {