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

go.sum

@@ -51,8 +51,6 @@ github.com/aliyun/alibaba-cloud-sdk-go v1.61.18/go.mod h1:v8ESoHo4SyHmuB4b1tJqDH
 github.com/antihax/optional v1.0.0/go.mod h1:uupD/76wgC+ih3iEmQUL+0Ugr19nfwCT1kdvxnR2qWY=
 github.com/apache/dubbo-getty v1.4.3 h1:PCKpryDasKOxwT5MBC6MIMO+0NLOaHF6Xco9YXQw7HI=
 github.com/apache/dubbo-getty v1.4.3/go.mod h1:ansXgKxxyhCOiQL29nO5ce1MDcEKmCyZuNR9oMs3hek=
-github.com/apache/dubbo-getty v1.4.4 h1:pthYQaCXyjHJ6/SjVwKkX5NhdAqSpUrRL1Z9GowrLdE=
-github.com/apache/dubbo-getty v1.4.4/go.mod h1:mcDyiu7M/TVrYDyL8TxDemQkOdvEqqHSQ4jOuYejY1w=
 github.com/apache/dubbo-go-hessian2 v1.9.1/go.mod h1:xQUjE7F8PX49nm80kChFvepA/AvqAZ0oh/UaB6+6pBE=
 github.com/apache/dubbo-go-hessian2 v1.9.2 h1:XuI8KvENSfKiAhiCBS4RNihmQDoPNmGWKT3gTui0p9A=
 github.com/apache/dubbo-go-hessian2 v1.9.2/go.mod h1:xQUjE7F8PX49nm80kChFvepA/AvqAZ0oh/UaB6+6pBE=
@@ -130,7 +128,6 @@ github.com/dubbogo/go-zookeeper v1.0.3 h1:UkuY+rBsxdT7Bs63QAzp9z7XqQ53W1j8E5rwl8
 github.com/dubbogo/go-zookeeper v1.0.3/go.mod h1:fn6n2CAEer3novYgk9ULLwAjuV8/g4DdC2ENwRb6E+c=
 github.com/dubbogo/gost v1.9.0/go.mod h1:pPTjVyoJan3aPxBPNUX0ADkXjPibLo+/Ib0/fADXSG8=
 github.com/dubbogo/gost v1.10.1/go.mod h1:+mQGS51XQEUWZP2JeGZTxJwipjRKtJO7Tr+FOg+72rI=
-github.com/dubbogo/gost v1.11.12/go.mod h1:vIcP9rqz2KsXHPjsAwIUtfJIJjppQLQDcYaZTy/61jI=
 github.com/dubbogo/gost v1.11.13 h1:sWvK1QbHpPBMmRQJV9qIH3syLegQBQa4xAPof3/Kv5c=
 github.com/dubbogo/gost v1.11.13/go.mod h1:vIcP9rqz2KsXHPjsAwIUtfJIJjppQLQDcYaZTy/61jI=
 github.com/dubbogo/jsonparser v1.0.1/go.mod h1:tYAtpctvSP/tWw4MeelsowSPgXQRVHHWbqL6ynps8jU=
@@ -430,7 +427,6 @@ github.com/modern-go/reflect2 v0.0.0-20180320133207-05fbef0ca5da/go.mod h1:bx2lN
 github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
 github.com/modern-go/reflect2 v1.0.1 h1:9f412s+6RmYXLWZSEzVVgPGK7C2PphHj5RJrvfx9AWI=
 github.com/modern-go/reflect2 v1.0.1/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
-github.com/montanaflynn/stats v0.6.6/go.mod h1:etXPPgVO6n31NxCd9KQUMvCM+ve0ruNzt6R8Bnaayow=
 github.com/mschoch/smat v0.2.0 h1:8imxQsjDm8yFEAVBe7azKmKSgzSkZXDuKkSq9374khM=
 github.com/mschoch/smat v0.2.0/go.mod h1:kc9mz7DoBKqDyiRL7VZN8KvXQMWeTaVnttLRXOlotKw=
 github.com/munnerz/goautoneg v0.0.0-20120707110453-a547fc61f48d/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=

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,7 +95,6 @@ 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)
 

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,6 @@ func newZkRegistry(url *common.URL) (registry.Registry, error) {
 		return nil, err
 	}
 
-	r.WaitGroup().Add(1) //zk client start successful, then wg +1
 	go zookeeper.HandleClientRestart(r)
 
 	r.listener = zookeeper.NewZkEventListener(r.client)
@@ -108,7 +107,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

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"
 )
 
@@ -46,55 +43,17 @@ type clientFacade interface {
 
 // HandleClientRestart keeps the connection between client and server
 func HandleClientRestart(r clientFacade) {
-	var (
-		err       error
-		failTimes int
-	)
-
+	r.WaitGroup().Add(1)
 	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

@@ -43,6 +43,7 @@ type ZkClientFacade interface {
 
 // HandleClientRestart keeps the connection between client and server
 func HandleClientRestart(r ZkClientFacade) {
+	r.WaitGroup().Add(1)
 	defer r.WaitGroup().Done()
 	for {
 		select {