本文基于Zookeeper和Curator实现一个简单的Key-Value配置中心，如果对Curator的使用尚不熟悉，可以参考上一篇博客《Zookeeper客户端Curator使用详解》。
设计思路：建立一个Zookeeper的持久化ZNode为根节点，根节点的路径为/configuration，然后每添加一个Key-Value项，在根节点下面添加一个子节点，节点的路径为/Key，节点的数据为Value的byte数组。
Paste_Image.png
代码实现相对简单：
ConfigurationCenter接口：
public interface ConfigurationCenter {
/**
* 添加配置
*
* @param key
* @param value
* @throws Exception e
*/
void addConfiguration(String key, String value) throws Exception;
/**
* 移除配置
*
* @param key
* @throws Exception e
*/
void deleteConfiguration(String key) throws Exception;
/**
* 更新配置
*
* @param key
* @throws Exception e
*/
void updateConfiguration(String key, String value) throws Exception;
/**
* 获取配置
*
* @param key
* @throws Exception e
*/
String getConfiguration(String key) throws Exception;
/**
* 获取所有配置
*
* @throws Exception e
*/
Map<String, String> getAllConfiguration() throws Exception;
}
实现类DefaultConfigurationCenter：
@Slf4j
public class DefaultConfigurationCenter implements ConfigurationCenter {
private static final String CONFIGURATION_ROOT_PATH = "/configuration";
private volatile Map<String, String> configurations = Maps.newHashMap();
private CuratorFramework client;
private PathChildrenCache pathChildrenCache;
private ExecutorService executor = Executors.newFixedThreadPool(2);
public DefaultConfigurationCenter(String connectionString) {
RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
this.client = CuratorFrameworkFactory.builder()
.connectString(connectionString)
.sessionTimeoutMs(50000)
.connectionTimeoutMs(50000)
.retryPolicy(retryPolicy)
.build();
initCuratorClient();
}
private void initCuratorClient() {
try {
client.start();
Stat stat = client.checkExists().forPath(CONFIGURATION_ROOT_PATH);
if (null == stat) {
client.create().withMode(CreateMode.PERSISTENT).forPath(CONFIGURATION_ROOT_PATH);
}
pathChildrenCache = new PathChildrenCache(client, CONFIGURATION_ROOT_PATH, true);
pathChildrenCache.getListenable().addListener((curatorFramework, event) -> {
String eventType = event.getType().name();
String path = event.getData().getPath();
String key = path.substring(path.lastIndexOf("/") + 1);
String data = null != event.getData() ? new String(event.getData().getData(), "UTF-8") : "";
log.debug(String.format("eventType:%s,path:%s,data:%s", eventType, path, data));
if (PathChildrenCacheEvent.Type.CHILD_ADDED == event.getType()
|| PathChildrenCacheEvent.Type.CHILD_UPDATED == event.getType()) {
reloadSingleConfigurationMapEntity(key, data);
} else if (PathChildrenCacheEvent.Type.CHILD_REMOVED == event.getType()) {
removeSingleConfigurationMapEntity(key);
}
}, executor);
pathChildrenCache.start();
initConfiguration();
} catch (Exception e) {
throw new RuntimeException(e);
}
}
private String concatKey(String key) {
return CONFIGURATION_ROOT_PATH.concat("/").concat(key);
}
private boolean checkChildPathExists(String key) throws Exception {
Stat stat = client.checkExists().forPath(concatKey(key));
if (null != stat) {
log.warn("configuration item {} exists!", key);
return true;
}
return false;
}
public void initConfiguration() throws Exception {
configurations = getAllConfiguration();
}
@Override
public void addConfiguration(String key, String value) throws Exception {
if (!checkChildPathExists(key)) {
client.create().withMode(CreateMode.PERSISTENT).forPath(concatKey(key), value.getBytes("UTF-8"));
}
}
@Override
public void deleteConfiguration(String key) throws Exception {
if (checkChildPathExists(key)) {
client.delete().forPath(concatKey(key));
} else {
log.warn("delete configuration item {} failed,item ZNode has not been found!", key);
}
}
@Override
public void updateConfiguration(String key, String value) throws Exception {
if (checkChildPathExists(key)) {
client.setData().forPath(concatKey(key), value.getBytes("UTF-8"));
} else {
log.warn("update configuration item {} failed,item ZNode has not been found!", key);
}
}
@Override
public String getConfiguration(String key) throws Exception {
if (configurations.containsKey(key)) {
return configurations.get(key);
}
if (checkChildPathExists(key)) {
return new String(client.getData().forPath(concatKey(key)), "UTF-8");
} else {
log.warn("get configuration item {} failed,item ZNode has not been found!", key);
return null;
}
}
private void reloadConfiguration() throws Exception {
List<String> childPaths = client.getChildren().forPath(CONFIGURATION_ROOT_PATH);
if (null != childPaths && !childPaths.isEmpty()) {
for (String childPath : childPaths) {
String key = childPath.substring(childPath.lastIndexOf("/") + 1);
String value = new String(client.getData().forPath(concatKey(childPath)));
reloadSingleConfigurationMapEntity(key, value);
}
}
}
@Override
public Map<String, String> getAllConfiguration() throws Exception {
if (!configurations.isEmpty()) {
return configurations;
}
List<String> childPaths = client.getChildren().forPath(CONFIGURATION_ROOT_PATH);
Map<String, String> configs = Maps.newHashMap();
if (null != childPaths && !childPaths.isEmpty()) {
for (String childPath : childPaths) {
String key = childPath.substring(childPath.indexOf("/") + 1);
String value = new String(client.getData().forPath(concatKey(childPath)), "UTF-8");
configs.put(key, value);
}
}
return configs;
}
private String reloadSingleConfigurationMapEntity(String key, String value) {
return configurations.put(key, value);
}
private String removeSingleConfigurationMapEntity(String key) {
if (configurations.containsKey(key)) {
return configurations.remove(key);
}
return null;
}
public void close() {
CloseableUtils.closeQuietly(pathChildrenCache);
CloseableUtils.closeQuietly(client);
if (null != executor && !executor.isShutdown()) {
executor.shutdown();
}
}
public static void main(String[] args) throws Exception {
String connectionInfo = "192.168.1.103:2181";
ConfigurationCenter configurationCenter = new DefaultConfigurationCenter(connectionInfo);
Map<String, String> all = configurationCenter.getAllConfiguration();
all.forEach((key, value) -> System.out.println("key:" + key + ",value:" + value));
configurationCenter.updateConfiguration("CORE_THREAD_SIEZ", "100");
configurationCenter.updateConfiguration("MAX_THREAD_SIEZ", "1000");
configurationCenter.updateConfiguration("QUEUE_SIEZ", "10000");
configurationCenter.updateConfiguration("CORE_THREAD_TIMEOUT_SECONDS", "600");
Thread.sleep(1000);
all = configurationCenter.getAllConfiguration();
all.forEach((key, value) -> System.out.println("key:" + key + ",value:" + value));
connectionInfo = "192.168.1.107:2181";
ConfigurationCenter configurationCenter2 = new DefaultConfigurationCenter(connectionInfo);
all = configurationCenter2.getAllConfiguration();
all.forEach((key, value) -> System.out.println("configurationCenter2 key:" + key + ",value:" + value));
connectionInfo = "192.168.1.101:2181";
ConfigurationCenter configurationCenter3 = new DefaultConfigurationCenter(connectionInfo);
all = configurationCenter3.getAllConfiguration();
all.forEach((key, value) -> System.out.println("configurationCenter3 key:" + key + ",value:" + value));
//测试获取无Znode节点的值
Thread.sleep(1000);
System.out.println("NONE_THIS_CONFIG : " + configurationCenter3.getConfiguration("NONE_THIS_CONFIG"));
}
}
可能代码和项目相对简陋，但是思路应该是没错的，仅供参考。
注意：
1.Value不一定必须指定为String类型，可以扩展为自定义的类型，但是需要自行实现序列化和反序列化；
2.Zookeeper单个节点数据越大，对网络方面的吞吐就会造成影响，所以节点数据越大读写性能越低，因此每个配置项的Value数据应该控制大小（但是对单个节点的操作不会因为节点总数增加而受到影响）。
项目仓库：
https://github.com/zjcscut/curator-seed
作图工具：
Xmind8
End on 2017-5-13 14:52.
Help yourselves!
我是throwable,在广州奋斗,白天上班,晚上和双休不定时加班,晚上有空坚持写下博客。
希望我的文章能够给你带来收获,共勉。
网友评论