如何實現一個分布式配置中心？
作者：小傅哥
配置中心在大廠系統開發中是一個非常常用的功能，它的核心功能在于不需要上線系統的情況下，改變系統中對象或者屬性的值。是屬性的值，也就是你在通過類獲取某個屬性，判斷；功能開關、渠道地址、人群名單、息費費率、切量占比等等，這些可能隨時動態調整的值，都是通過配置中心實現的。
本文的宗旨在于通過簡單干凈實踐的方式，向讀者介紹 Zookeeper 的安裝配置，學習 SpringBoot 整合使用，以及基于 Zookeeper 開發一個簡單的配置中心功能內核。通過這樣的實踐方式，讓讀者入門和掌握 Zookeeper 以應對后續需要此技術棧的相關開發項目。
本文的重點是基于 Zookeeper 實現的配置中心，那配置中心是啥呢？
配置中心在大廠系統開發中是一個非常常用的功能，它的核心功能在于不需要上線系統的情況下，改變系統中對象或者屬性的值。是屬性的值，也就是你在通過類獲取某個屬性，判斷；功能開關、渠道地址、人群名單、息費費率、切量占比等等，這些可能隨時動態調整的值，都是通過配置中心實現的。所以在本章節的案例中，小傅哥基于 Zookeeper 組件的功能特性，來設計這樣一個配置中心，方便大家學習。
本文涉及的工程：
- xfg-dev-tech-zookeeper：https://gitcode.net/KnowledgePlanet/road-map/xfg-dev-tech-connection-pool - docs/dev-ops 提供了 Zookeeper 安裝腳本。
一、配置中心
Zookeeper 有什么特性，動態配置中心怎么做？
技術是支撐解決方案實現的，有了各個技術棧組件的自身特點，才好實現出我們所需的各類功能。那么這樣的一個能讓，各個服務都可以動態變更配置的配置中心，就要用到 Zookeeper 的節點監聽和節點值的變化來動態設置 Java 類中屬性的變化。如圖：
圖片
- 首先，我們需要定義出一個 Zookeeper 監聽的配置路徑，一般這個路徑在配置中心中是申請的系統使用地址，以確保值的唯一。
- 之后，每個類對應的屬性，需要映射出一個監聽的節點。比如；Zookeeper 監聽了 /xfg-dev-tech/config 那么類中 a 屬性可以是 /xfg-dev-tech/config/a 這對這個路徑設置的值，就可以被監聽拿到了。
- 最后，把獲取到的監聽值，通過 Java 反射操作，把值設置到對應的屬性上。這樣在 SpringBoot 應用程序中，使用某個類的屬性值的時候，就可以動態的獲取到變化的屬性值了。
二、環境配置
在安裝執行 docker-compose.yml 腳本之前，你需要先在本地安裝 docker 之后 IntelliJ IDEA 打開 docker-compose.yml 文件，如圖操作即可安裝。
圖片
圖片
- 另外，如果你是在服務器上安裝，則需要執行 docker-compose -f docker-compose.yml up -d 并且是你已經安裝了 Docker-Compose 包?！?這些內容在小傅哥的《Java 簡明教程》中都有講解，可以進入學習。
三、基本使用
連接腳本：
docker exec -it zookeeper bash
zkCli.sh -server IP(替換為你自己的):2181常用命令：
1. 創建節點：create /path data
2. 創建臨時節點：create -e /path data
3. 創建順序節點：create -s /path data
4. 創建臨時順序節點：create -e -s /path data
5. 獲取節點數據：get /path
6. 獲取節點子節點列表：ls /path
7. 更新節點數據：set /path data
8. 刪除節點：delete /path
9. 刪除節點及其子節點：deleteall /path
10. 監聽節點變化：get -w /path
11. 查看節點狀態：stat /path
12. 查看節點ACL權限：getAcl /path
13. 設置節點ACL權限：setAcl /path acl
14. 查看節點子節點數量：count /path
15. 查看節點子節點數量并監聽變化：count -w /pathroot@4365b68d50d6:/apache-zookeeper-3.9.0-bin# ls
bin conf docs lib LICENSE.txt NOTICE.txt README.md README_packaging.md
root@4365b68d50d6:/apache-zookeeper-3.9.0-bin# zkCli.sh -server 10.253.6.71:2181
[zk: 192.168.1.101:2181(CONNECTED) 1] ls /xfg-dev-tech
[config, configdowngradeSwitch]
[zk: 192.168.1.101:2181(CONNECTED) 2]執行完鏈接 Zookeeper 以后，就可以執行這些常用命令了。你也可以嘗試著練習下這些命令。
四、功能實現
1. 工程結構
圖片
工程結構分為2個部分：
- app 啟動層的 config 包下，用于提供 Zookeeper 服務的啟動配置。以及小傅哥在這里新添加的功能 DCCValue 配置中心模塊。
- trigger 是觸發器，這里吧 http 請求、listener 監聽，都是放到這里使用。另外像 MQ、JOB、RPC 也是放到這一層，以這一層觸發，來調用我們的領域服務。
2. 啟動 Zookeeper 服務
2.1 自定配置
@Data
@ConfigurationProperties(prefix = "zookeeper.sdk.config", ignoreInvalidFields = true)
public class ZookeeperClientConfigProperties {
private String connectString;
private int baseSleepTimeMs;
private int maxRetries;
private int sessionTimeoutMs;
private int connectionTimeoutMs;
}2.2 使用配置
zookeeper:
sdk:
config:
connect-string: 10.253.6.71:2181
base-sleep-time-ms: 1000
max-retries: 3
session-timeout-ms: 1800000
connection-timeout-ms: 300002.3 配置服務
@Configuration
@EnableConfigurationProperties(ZookeeperClientConfigProperties.class)
public class ZooKeeperClientConfig {
/**
* 多參數構建ZooKeeper客戶端連接
*
* @return client
*/
@Bean(name = "zookeeperClient")
public CuratorFramework createWithOptions(ZookeeperClientConfigProperties properties) {
ExponentialBackoffRetry backoffRetry = new ExponentialBackoffRetry(properties.getBaseSleepTimeMs(), properties.getMaxRetries());
CuratorFramework client = CuratorFrameworkFactory.builder()
.connectString(properties.getConnectString())
.retryPolicy(backoffRetry)
.sessionTimeoutMs(properties.getSessionTimeoutMs())
.connectionTimeoutMs(properties.getConnectionTimeoutMs())
.build();
client.start();
return client;
}
}- 這樣我們就可以啟動一個 Zookeeper 的客戶端了，自定義可以更好的控制和使用。
3. 定義注解
就功能來講，我們需要對類中的屬性進行賦值操作。那么就需要使用自定義注解進行標記。所以這里我們先自定義一個注解。
@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.FIELD})
@Documented
public @interface DCCValue {
String value() default "";
}- 這樣所有使用了 @DCCValue 的注解的字段就都可以被我掃描到了。
4. 監聽變化
4.1 獲取屬性
源碼：cn.bugstack.xfg.dev.tech.config.DCCValueBeanFactory#postProcessAfterInitialization
@Override
public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
Class<?> beanClass = bean.getClass();
Field[] fields = beanClass.getDeclaredFields();
for (Field field : fields) {
if (field.isAnnotationPresent(DCCValue.class)) {
DCCValue dccValue = field.getAnnotation(DCCValue.class);
try {
if (null == client.checkExists().forPath(BASE_CONFIG_PATH.concat("/").concat(dccValue.value()))) {
client.create().creatingParentsIfNeeded().forPath(BASE_CONFIG_PATH.concat("/").concat(dccValue.value()));
log.info("DCC 節點監聽 listener node {} not absent create new done!", BASE_CONFIG_PATH.concat("/").concat(dccValue.value()));
}
} catch (Exception e) {
throw new RuntimeException(e);
}
dccObjGroup.put(BASE_CONFIG_PATH.concat("/").concat(dccValue.value()), bean);
}
}
return bean;
}- DCCValueBeanFactory 實現了 BeanPostProcessor 接口的 postProcessAfterInitialization 方法。
- 在實現中，通過對 bean 對象的解析獲取到使用 DCCValue 注解屬性，并判斷這個屬性拼接的地址是否在 Zookeeper 中創建，如果沒有則創建。之后保存對象到內存中。
4.2 設置屬性
源碼：cn.bugstack.xfg.dev.tech.config.DCCValueBeanFactory#DCCValueBeanFactory
curatorCache.listenable().addListener((type, oldData, data) -> {
switch (type) {
case NODE_CHANGED:
String dccValuePath = data.getPath();
Object objBean = dccObjGroup.get(dccValuePath);
try {
// 1. getDeclaredField 方法用于獲取指定類中聲明的所有字段，包括私有字段、受保護字段和公共字段。
// 2. getField 方法用于獲取指定類中的公共字段，即只能獲取到公共訪問修飾符（public）的字段。
Field field = objBean.getClass().getDeclaredField(dccValuePath.substring(dccValuePath.lastIndexOf("/") + 1));
field.setAccessible(true);
field.set(objBean, new String(data.getData()));
field.setAccessible(false);
} catch (Exception e) {
throw new RuntimeException(e);
}
break;
default:
break;
}
});- 基于 Zookeeper 對節點的監聽，只要這個節點上有值發生變化。就可以立刻檢測到對應的路徑信息和值信息。
- 那么拿到這個值信息，就可以把值寫入到對應的屬性上了。如類.A = Zookeeper 獲取到的值
五、功能使用
源碼：cn.bugstack.xfg.dev.tech.trigger.http.ConfigController
@RestController
public class ConfigController {
@DCCValue("downgradeSwitch")
private String downgradeSwitch;
@DCCValue("userWhiteList")
private String userWhiteList;
@Resource
private CuratorFramework curatorFramework;
/**
* curl http://localhost:8091/getConfig/downgradeSwitch
*/
@RequestMapping("/getConfig/downgradeSwitch")
public String getConfigDowngradeSwitch() {
return downgradeSwitch;
}
/**
* curl http://localhost:8091/getConfig/userWhiteList
*/
@RequestMapping("/getConfig/userWhiteList")
public String getConfigUserWhiteList() {
return userWhiteList;
}
/**
* curl -X GET "http://localhost:8091/setConfig?downgradeSwitch=false&userWhiteList=xfg,user2,user3"
*/
@GetMapping("/setConfig")
public void setConfig(Boolean downgradeSwitch, String userWhiteList) throws Exception {
curatorFramework.setData().forPath("/xfg-dev-tech/config/downgradeSwitch", (downgradeSwitch ? "開" : "關").getBytes(StandardCharsets.UTF_8));
curatorFramework.setData().forPath("/xfg-dev-tech/config/userWhiteList", userWhiteList.getBytes(StandardCharsets.UTF_8));
}
}這里的核心驗證就是讓 downgradeSwitch、userWhiteList 這2個屬性值可以動態變化；
- 在兩個屬性上添加注解后，就會被掃描和管理。
- 獲取值方法：http://localhost:8091/getConfig/downgradeSwitch、http://localhost:8091/getConfig/userWhiteList
- 設置值方法：http://localhost:8091/setConfig?downgradeSwitch=false&userWhiteList=xfg,user2,user3 - 這里的設置值操作不非得在這里，可以是一個單獨的控制后臺來操作。這里的方式主要是演示作用
圖片
你可以按照如圖的操作順序，進行驗證屬性值的變化。
六、其他測試
@Slf4j
@RunWith(SpringRunner.class)
@SpringBootTest
public class ApiTest {
@Resource
private CuratorFramework curatorFramework;
@Test
public void test_all() throws Exception {
String path = "/xfg-dev-tech/config/downgradeSwitch";
String data = "0";
curatorFramework.create().withMode(CreateMode.EPHEMERAL).forPath(path, data.getBytes(StandardCharsets.UTF_8));
for (int i = 0; i < 2; i++) {
curatorFramework.setData().forPath(path, String.valueOf(i).getBytes(StandardCharsets.UTF_8));
}
}
/**
* 創建永久節點
*/
@Test
public void createNode() throws Exception {
String path = "/xfg-dev-tech/config/downgradeSwitch/test/a";
String data = "0";
if (null == curatorFramework.checkExists().forPath(path)) {
curatorFramework.create().creatingParentsIfNeeded().forPath(path);
}
}
/**
* 創建臨時節點
*/
@Test
public void createEphemeralNode() throws Exception {
String path = "/xfg-dev-tech/config/epnode";
String data = "0";
curatorFramework.create().withMode(CreateMode.EPHEMERAL).forPath(path, data.getBytes(StandardCharsets.UTF_8));
}
/**
* 創建臨時有序節點
*/
@Test
public void crateEphemeralSequentialNode() throws Exception {
String path = "/xfg-dev-tech/config/epsnode";
String data = "0";
curatorFramework.create()
.withMode(CreateMode.EPHEMERAL_SEQUENTIAL)
.forPath(path, data.getBytes(StandardCharsets.UTF_8));
}
/**
* 往節點種設置數據
*/
@Test
public void setData() throws Exception {
curatorFramework.setData().forPath("/xfg-dev-tech/config/downgradeSwitch", "111".getBytes(StandardCharsets.UTF_8));
curatorFramework.setData().forPath("/xfg-dev-tech/config/userWhiteList", "222".getBytes(StandardCharsets.UTF_8));
}
@Test
public void getData() throws Exception {
String downgradeSwitch = new String(curatorFramework.getData().forPath("/xfg-dev-tech/config/downgradeSwitch"), StandardCharsets.UTF_8);
log.info("測試結果: {}", downgradeSwitch);
String userWhiteList = new String(curatorFramework.getData().forPath("/xfg-dev-tech/config/userWhiteList"), StandardCharsets.UTF_8);
log.info("測試結果: {}", userWhiteList);
}
/**
* 異步修改數據
*/
@Test
public void setDataAsync() throws Exception {
String path = "/xfg-dev-tech/config/downgradeSwitch";
String data = "0";
CuratorListener listener = (client, event) -> {
Stat stat = event.getStat();
log.info("stat=" + JSON.toJSONString(stat));
CuratorEventType eventType = event.getType();
log.info("eventType=" + eventType.name());
};
curatorFramework.getCuratorListenable().addListener(listener);
curatorFramework.setData().inBackground().forPath(path, data.getBytes(StandardCharsets.UTF_8));
}
/**
* 刪除節點
*/
@Test
public void deleteData() throws Exception {
String path = "/xfg-dev-tech/config/downgradeSwitch";
curatorFramework.delete().deletingChildrenIfNeeded().forPath(path);
}
/**
* 安全刪除節點
*/
@Test
public void guaranteedDeleteData() throws Exception {
String path = "/xfg-dev-tech/config/downgradeSwitch";
curatorFramework.delete().guaranteed().forPath(path);
}
/**
* 獲取子節點下的全部子節點路徑集合
*/
@Test
public void watchedGetChildren() throws Exception {
String path = "/xfg-dev-tech";
List<String> children = curatorFramework.getChildren().watched().forPath(path);
log.info("測試結果：{}", JSON.toJSONString(children));
}
/**
* 獲取節點數據
*/
@Test
public void getDataByPath() throws Exception {
String path = "/xfg-dev-tech/config/downgradeSwitch";
String fullClassName = "";
String jsonStr = new String(curatorFramework.getData().forPath(path), StandardCharsets.UTF_8);
Class clazz = Class.forName(fullClassName);
log.info("測試結果：{}", JSON.parseObject(jsonStr, clazz));
}
}- 這些功能也都可以測試驗證，也是平常用的較多的東西。
七、其他資料
- Zookeeper Web UI：https://zoonavigator.elkozmon.com/en/latest/
- 官網文檔：https://zookeeper.apache.org/doc/r3.5.0-alpha/zookeeperAdmin.html
責任編輯：武曉燕
來源：
bugstack蟲洞棧
