Skip to content

Commit 9598cd0

Browse files
cvictorychickenlj
cvictory
authored and
chickenlj
committed
Merge pull request #3603, configcenter share zookeeper connection with registry.
Fixes #3288
1 parent 0c2232f commit 9598cd0

File tree

21 files changed

+396
-748
lines changed

21 files changed

+396
-748
lines changed

‎dubbo-configcenter/dubbo-configcenter-api/src/main/java/org/apache/dubbo/configcenter/ConfigChangeEvent.java‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -49,4 +49,12 @@ public ConfigChangeType getChangeType() {
4949
return changeType;
5050
}
5151

52-
}
52+
@Override
53+
public String toString() {
54+
return "ConfigChangeEvent{" +
55+
"key='" + key + '\'' +
56+
", value='" + value + '\'' +
57+
", changeType=" + changeType +
58+
'}';
59+
}
60+
}

‎dubbo-configcenter/dubbo-configcenter-zookeeper/pom.xml‎

Lines changed: 4 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -33,21 +33,14 @@
3333
<version>${project.parent.version}</version>
3434
</dependency>
3535
<dependency>
36-
<groupId>org.apache.curator</groupId>
37-
<artifactId>curator-framework</artifactId>
38-
</dependency>
39-
<dependency>
40-
<groupId>org.apache.curator</groupId>
41-
<artifactId>curator-recipes</artifactId>
42-
</dependency>
43-
<dependency>
44-
<groupId>org.apache.zookeeper</groupId>
45-
<artifactId>zookeeper</artifactId>
36+
<groupId>org.apache.dubbo</groupId>
37+
<artifactId>dubbo-remoting-zookeeper</artifactId>
38+
<version>${project.parent.version}</version>
4639
</dependency>
4740
<dependency>
4841
<groupId>org.apache.curator</groupId>
4942
<artifactId>curator-test</artifactId>
5043
<scope>test</scope>
5144
</dependency>
5245
</dependencies>
53-
</project>
46+
</project>

‎dubbo-configcenter/dubbo-configcenter-zookeeper/src/main/java/org/apache/dubbo/configcenter/support/zookeeper/CacheListener.java‎

Lines changed: 43 additions & 51 deletions
Original file line numberDiff line numberDiff line change
@@ -21,13 +21,9 @@
2121
import org.apache.dubbo.configcenter.ConfigChangeEvent;
2222
import org.apache.dubbo.configcenter.ConfigChangeType;
2323
import org.apache.dubbo.configcenter.ConfigurationListener;
24+
import org.apache.dubbo.remoting.zookeeper.DataListener;
25+
import org.apache.dubbo.remoting.zookeeper.EventType;
2426

25-
import org.apache.curator.framework.CuratorFramework;
26-
import org.apache.curator.framework.recipes.cache.ChildData;
27-
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
28-
import org.apache.curator.framework.recipes.cache.TreeCacheListener;
29-
30-
import java.nio.charset.StandardCharsets;
3127
import java.util.Map;
3228
import java.util.Set;
3329
import java.util.concurrent.ConcurrentHashMap;
@@ -37,9 +33,8 @@
3733
/**
3834
*
3935
*/
40-
public class CacheListener implements TreeCacheListener {
41-
private static final byte[] EMPTY_BYTES = new byte[0];
4236

37+
public class CacheListener implements DataListener {
4338
private Map<String, Set<ConfigurationListener>> keyListeners = new ConcurrentHashMap<>();
4439
private CountDownLatch initializedLatch;
4540
private String rootPath;
@@ -49,76 +44,73 @@ public CacheListener(String rootPath, CountDownLatch initializedLatch) {
4944
this.initializedLatch = initializedLatch;
5045
}
5146

47+
public void addListener(String key, ConfigurationListener configurationListener) {
48+
Set<ConfigurationListener> listeners = this.keyListeners.computeIfAbsent(key, k -> new CopyOnWriteArraySet<>());
49+
listeners.add(configurationListener);
50+
}
51+
52+
public void removeListener(String key, ConfigurationListener configurationListener) {
53+
Set<ConfigurationListener> listeners = this.keyListeners.get(key);
54+
if (listeners != null) {
55+
listeners.remove(configurationListener);
56+
}
57+
}
58+
59+
/**
60+
* This is used to convert a configuration nodePath into a key
61+
* TODO doc
62+
*
63+
* @param path
64+
* @return key (nodePath less the config root path)
65+
*/
66+
private String pathToKey(String path) {
67+
if (StringUtils.isEmpty(path)) {
68+
return path;
69+
}
70+
return path.replace(rootPath + "/", "").replaceAll("/", ".");
71+
}
72+
73+
5274
@Override
53-
public void childEvent(CuratorFramework aClient, TreeCacheEvent event) throws Exception {
75+
public void dataChanged(String path, Object value, EventType eventType) {
76+
if (eventType == null) {
77+
return;
78+
}
5479

55-
TreeCacheEvent.Type type = event.getType();
56-
ChildData data = event.getData();
57-
if (type == TreeCacheEvent.Type.INITIALIZED) {
80+
if (eventType == EventType.INITIALIZED) {
5881
initializedLatch.countDown();
5982
return;
6083
}
6184

62-
// TODO, ignore other event types
63-
if (data == null) {
85+
if (path == null || (value == null && eventType != EventType.NodeDeleted)) {
6486
return;
6587
}
6688

6789
// TODO We limit the notification of config changes to a specific path level, for example
6890
// /dubbo/config/service/configurators, other config changes not in this level will not get notified,
6991
// say /dubbo/config/dubbo.properties
70-
if (data.getPath().split("/").length >= 5) {
71-
byte[] value = data.getData();
72-
String key = pathToKey(data.getPath());
92+
if (path.split("/").length >= 5) {
93+
String key = pathToKey(path);
7394
ConfigChangeType changeType;
74-
switch (type) {
75-
case NODE_ADDED:
95+
switch (eventType) {
96+
case NodeCreated:
7697
changeType = ConfigChangeType.ADDED;
7798
break;
78-
case NODE_REMOVED:
99+
case NodeDeleted:
79100
changeType = ConfigChangeType.DELETED;
80101
break;
81-
case NODE_UPDATED:
102+
case NodeDataChanged:
82103
changeType = ConfigChangeType.MODIFIED;
83104
break;
84105
default:
85106
return;
86107
}
87108

88-
if (value == null) {
89-
value = EMPTY_BYTES;
90-
}
91-
ConfigChangeEvent configChangeEvent = new ConfigChangeEvent(key, new String(value, StandardCharsets.UTF_8), changeType);
109+
ConfigChangeEvent configChangeEvent = new ConfigChangeEvent(key, (String) value, changeType);
92110
Set<ConfigurationListener> listeners = keyListeners.get(key);
93111
if (CollectionUtils.isNotEmpty(listeners)) {
94112
listeners.forEach(listener -> listener.process(configChangeEvent));
95113
}
96114
}
97115
}
98-
99-
public void addListener(String key, ConfigurationListener configurationListener) {
100-
Set<ConfigurationListener> listeners = this.keyListeners.computeIfAbsent(key, k -> new CopyOnWriteArraySet<>());
101-
listeners.add(configurationListener);
102-
}
103-
104-
public void removeListener(String key, ConfigurationListener configurationListener) {
105-
Set<ConfigurationListener> listeners = this.keyListeners.get(key);
106-
if (listeners != null) {
107-
listeners.remove(configurationListener);
108-
}
109-
}
110-
111-
/**
112-
* This is used to convert a configuration nodePath into a key
113-
* TODO doc
114-
*
115-
* @param path
116-
* @return key (nodePath less the config root path)
117-
*/
118-
private String pathToKey(String path) {
119-
if (StringUtils.isEmpty(path)) {
120-
return path;
121-
}
122-
return path.replace(rootPath + "/", "").replaceAll("/", ".");
123-
}
124116
}

‎dubbo-configcenter/dubbo-configcenter-zookeeper/src/main/java/org/apache/dubbo/configcenter/support/zookeeper/ZookeeperDynamicConfiguration.java‎

Lines changed: 13 additions & 58 deletions
Original file line numberDiff line numberDiff line change
@@ -16,81 +16,54 @@
1616
*/
1717
package org.apache.dubbo.configcenter.support.zookeeper;
1818

19-
import org.apache.dubbo.common.Constants;
2019
import org.apache.dubbo.common.URL;
2120
import org.apache.dubbo.common.utils.NamedThreadFactory;
2221
import org.apache.dubbo.common.utils.StringUtils;
2322
import org.apache.dubbo.configcenter.ConfigurationListener;
2423
import org.apache.dubbo.configcenter.DynamicConfiguration;
24+
import org.apache.dubbo.remoting.zookeeper.ZookeeperClient;
25+
import org.apache.dubbo.remoting.zookeeper.ZookeeperTransporter;
2526

26-
import org.apache.curator.RetryPolicy;
27-
import org.apache.curator.framework.CuratorFramework;
28-
import org.apache.curator.framework.recipes.cache.ChildData;
29-
import org.apache.curator.framework.recipes.cache.TreeCache;
30-
import org.apache.curator.retry.ExponentialBackoffRetry;
3127
import org.slf4j.Logger;
3228
import org.slf4j.LoggerFactory;
3329

34-
import java.nio.charset.StandardCharsets;
3530
import java.util.concurrent.CountDownLatch;
3631
import java.util.concurrent.Executor;
3732
import java.util.concurrent.Executors;
38-
import java.util.concurrent.TimeUnit;
3933

40-
import static org.apache.curator.framework.CuratorFrameworkFactory.newClient;
4134
import static org.apache.dubbo.common.Constants.CONFIG_NAMESPACE_KEY;
4235

4336
/**
4437
*
4538
*/
4639
public class ZookeeperDynamicConfiguration implements DynamicConfiguration {
4740
private static final Logger logger = LoggerFactory.getLogger(ZookeeperDynamicConfiguration.class);
48-
private Executor executor;
49-
private CuratorFramework client;
5041

42+
private Executor executor;
5143
// The final root path would be: /configRootPath/"config"
5244
private String rootPath;
53-
private TreeCache treeCache;
45+
private final ZookeeperClient zkClient;
5446
private CountDownLatch initializedLatch;
5547

5648
private CacheListener cacheListener;
5749
private URL url;
5850

5951

60-
ZookeeperDynamicConfiguration(URL url) {
52+
ZookeeperDynamicConfiguration(URL url, ZookeeperTransporter zookeeperTransporter) {
6153
this.url = url;
6254
rootPath = "/" + url.getParameter(CONFIG_NAMESPACE_KEY, DEFAULT_GROUP) + "/config";
6355

64-
RetryPolicy policy = new ExponentialBackoffRetry(1000, 3);
65-
int sessionTimeout = url.getParameter("config.session.timeout", 60 * 1000);
66-
int connectTimeout = url.getParameter("config.connect.timeout", 10 * 1000);
67-
String connectString = url.getBackupAddress();
68-
client = newClient(connectString, sessionTimeout, connectTimeout, policy);
69-
client.start();
70-
71-
try {
72-
boolean connected = client.blockUntilConnected(3 * connectTimeout, TimeUnit.MILLISECONDS);
73-
if (!connected) {
74-
if (url.getParameter(Constants.CONFIG_CHECK_KEY, true)) {
75-
throw new IllegalStateException("Failed to connect to config center (zookeeper): "
76-
+ connectString + " in " + 3 * connectTimeout + "ms.");
77-
} else {
78-
logger.warn("The config center (zookeeper) is not fully initialized in " + 3 * connectTimeout + "ms, address is: " + connectString);
79-
}
80-
}
81-
} catch (InterruptedException e) {
82-
throw new IllegalStateException("The thread was interrupted unexpectedly when trying connecting to zookeeper "
83-
+ connectString + " config center, ", e);
84-
}
85-
8656
initializedLatch = new CountDownLatch(1);
8757
this.cacheListener = new CacheListener(rootPath, initializedLatch);
8858
this.executor = Executors.newFixedThreadPool(1, new NamedThreadFactory(this.getClass().getSimpleName(), true));
89-
// build local cache
59+
60+
zkClient = zookeeperTransporter.connect(url);
61+
zkClient.addDataListener(rootPath, cacheListener, executor);
9062
try {
91-
this.buildCache();
92-
} catch (Exception e) {
93-
logger.warn("Failed to build local cache for config center (zookeeper), address is ." + connectString);
63+
// Wait for connection
64+
this.initializedLatch.await();
65+
} catch (InterruptedException e) {
66+
logger.warn("Failed to build local cache for config center (zookeeper)." + url);
9467
}
9568
}
9669

@@ -100,11 +73,7 @@ public class ZookeeperDynamicConfiguration implements DynamicConfiguration {
10073
*/
10174
@Override
10275
public Object getInternalProperty(String key) {
103-
ChildData childData = treeCache.getCurrentData(key);
104-
if (childData != null) {
105-
return new String(childData.getData(), StandardCharsets.UTF_8);
106-
}
107-
return null;
76+
return zkClient.getContent(key);
10877
}
10978

11079
/**
@@ -141,18 +110,4 @@ public String getConfig(String key, String group, long timeout) throws IllegalSt
141110

142111
return (String) getInternalProperty(rootPath + "/" + key);
143112
}
144-
145-
/**
146-
* Adds a listener to the pathChildrenCache, initializes the cache, then starts the cache-management background
147-
* thread
148-
*/
149-
private void buildCache() throws Exception {
150-
this.treeCache = new TreeCache(client, rootPath);
151-
// create the watcher for future configuration updates
152-
treeCache.getListenable().addListener(cacheListener, executor);
153-
154-
// it's not blocking, so we use an extra latch 'initializedLatch' to make sure cache fully initialized before use.
155-
treeCache.start();
156-
initializedLatch.await();
157-
}
158113
}

‎dubbo-configcenter/dubbo-configcenter-zookeeper/src/main/java/org/apache/dubbo/configcenter/support/zookeeper/ZookeeperDynamicConfigurationFactory.java‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,13 +19,22 @@
1919
import org.apache.dubbo.common.URL;
2020
import org.apache.dubbo.configcenter.AbstractDynamicConfigurationFactory;
2121
import org.apache.dubbo.configcenter.DynamicConfiguration;
22+
import org.apache.dubbo.remoting.zookeeper.ZookeeperTransporter;
2223

2324
/**
2425
*
2526
*/
2627
public class ZookeeperDynamicConfigurationFactory extends AbstractDynamicConfigurationFactory {
28+
29+
private ZookeeperTransporter zookeeperTransporter;
30+
31+
public void setZookeeperTransporter(ZookeeperTransporter zookeeperTransporter) {
32+
this.zookeeperTransporter = zookeeperTransporter;
33+
}
34+
35+
2736
@Override
2837
protected DynamicConfiguration createDynamicConfiguration(URL url) {
29-
return new ZookeeperDynamicConfiguration(url);
38+
return new ZookeeperDynamicConfiguration(url, zookeeperTransporter);
3039
}
3140
}

‎dubbo-configcenter/dubbo-configcenter-zookeeper/src/test/java/org/apache/dubbo/configcenter/support/zookeeper/ZookeeperDynamicConfigurationTest.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -133,6 +133,7 @@ public TestListener(CountDownLatch latch) {
133133

134134
@Override
135135
public void process(ConfigChangeEvent event) {
136+
System.out.println(this + ": " + event);
136137
Integer count = countMap.computeIfAbsent(event.getKey(), k -> new Integer(0));
137138
countMap.put(event.getKey(), ++count);
138139

‎dubbo-dependencies-bom/pom.xml‎

Lines changed: 1 addition & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -101,7 +101,6 @@
101101
<httpcore_version>4.4.6</httpcore_version>
102102
<fastjson_version>1.2.46</fastjson_version>
103103
<zookeeper_version>3.4.13</zookeeper_version>
104-
<zkclient_version>0.2</zkclient_version>
105104
<curator_version>4.0.1</curator_version>
106105
<curator_test_version>2.12.0</curator_test_version>
107106
<jedis_version>2.9.0</jedis_version>
@@ -151,7 +150,7 @@
151150
<dependencies>
152151
<!-- Common libs -->
153152
<dependency>
154-
<groupId>org.springframework</groupId>
153+
<groupId>org.springframework</groupId>
155154
<artifactId>spring-framework-bom</artifactId>
156155
<version>${spring_version}</version>
157156
<type>pom</type>
@@ -208,17 +207,6 @@
208207
</exclusion>
209208
</exclusions>
210209
</dependency>
211-
<dependency>
212-
<groupId>com.101tec</groupId>
213-
<artifactId>zkclient</artifactId>
214-
<version>${zkclient_version}</version>
215-
<exclusions>
216-
<exclusion>
217-
<groupId>org.apache.zookeeper</groupId>
218-
<artifactId>zookeeper</artifactId>
219-
</exclusion>
220-
</exclusions>
221-
</dependency>
222210
<dependency>
223211
<groupId>org.apache.curator</groupId>
224212
<artifactId>curator-framework</artifactId>

‎dubbo-dependencies/dubbo-dependencies-zookeeper/pom.xml‎

Lines changed: 0 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -50,10 +50,6 @@
5050
<groupId>org.apache.curator</groupId>
5151
<artifactId>curator-recipes</artifactId>
5252
</dependency>
53-
<dependency>
54-
<groupId>com.101tec</groupId>
55-
<artifactId>zkclient</artifactId>
56-
</dependency>
5753
<dependency>
5854
<groupId>org.apache.zookeeper</groupId>
5955
<artifactId>zookeeper</artifactId>

0 commit comments

Comments
 (0)
⚡