This is an automated email from the ASF dual-hosted git repository.
laurence pushed a commit to branch 3.0
in repository https://gitbox.apache.org/repos/asf/dubbo-go.git
The following commit(s) were added to refs/heads/3.0 by this push:
new fb0d226 fix:#1143 Feature/reduce etcd registry conn; wait group
modify (#1297)
fb0d226 is described below
commit fb0d2266be6338f27d7a505bd4f653be98e21c97
Author: WilliamLeaves <[email protected]>
AuthorDate: Thu Jul 29 22:38:18 2021 +0800
fix:#1143 Feature/reduce etcd registry conn; wait group modify (#1297)
* modify map use in nacos service_discovery
* fix map double write bug
* modify etcd registry restart mechanism
* add sync.once for HandleClientRestart
* modify the logic when add() or done() of wait group in zk,etcd and k8s
registry
* modify the logic when add() or done() of wait group in zk,etcd and k8s
registry
* add handleClientRestart func
* add handleClientRestart func
* go fmt
* fix: update linter
Co-authored-by: yexiaobo <[email protected]>
Co-authored-by: LaurenceLiZhixin <[email protected]>
---
.github/workflows/github-actions.yml | 2 +-
registry/base_registry.go | 6 ++---
registry/etcdv3/registry.go | 9 ++++---
registry/kubernetes/registry.go | 5 ++--
registry/zookeeper/registry.go | 8 ++++--
remoting/etcdv3/facade.go | 52 +++++-------------------------------
remoting/zookeeper/facade.go | 1 +
7 files changed, 25 insertions(+), 58 deletions(-)
diff --git a/.github/workflows/github-actions.yml
b/.github/workflows/github-actions.yml
index 952e58d..413a39f 100644
--- a/.github/workflows/github-actions.yml
+++ b/.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
diff --git a/registry/base_registry.go b/registry/base_registry.go
index dec2786..31d5b3f 100644
--- a/registry/base_registry.go
+++ b/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
diff --git a/registry/etcdv3/registry.go b/registry/etcdv3/registry.go
index 6be5015..b6e4a15 100644
--- a/registry/etcdv3/registry.go
+++ b/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)
+}
diff --git a/registry/kubernetes/registry.go b/registry/kubernetes/registry.go
index faec59d..239644e 100644
--- a/registry/kubernetes/registry.go
+++ b/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 {
diff --git a/registry/zookeeper/registry.go b/registry/zookeeper/registry.go
index afe9772..ea7e342 100644
--- a/registry/zookeeper/registry.go
+++ b/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)
+}
diff --git a/remoting/etcdv3/facade.go b/remoting/etcdv3/facade.go
index 496f1c7..a5f3b9e 100644
--- a/remoting/etcdv3/facade.go
+++ b/remoting/etcdv3/facade.go
@@ -25,12 +25,10 @@ import (
import (
getty "github.com/apache/dubbo-getty"
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 +43,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
}
}
}
diff --git a/remoting/zookeeper/facade.go b/remoting/zookeeper/facade.go
index aa96232..f9a4d59 100644
--- a/remoting/zookeeper/facade.go
+++ b/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 {
