上一篇（基于zookeeper實現分布式配置中心（一））講述了zookeeper相關概念和工作原理。接下來根據zookeeper的特性，簡單實現一個分布式配置中心。

配置中心的優勢

1、各環境配置集中管理。

2、配置更改，實時推送，jvm環境變量及時生效。

3、依靠配置變更，動態擴展功能，減少二次上線帶來的成本。

4、減少開發人員、運維人員修改配置帶來的額外開銷。

?配置中心架構圖

　　

配置中心功能

1、配置管理平臺中，操作人員可以創建項目所屬系統、應用名稱、實例名稱、配置分組等信息。

2、配置管理平臺中，操作人員可以上傳配置文件，對屬性有增、刪、改、查的操作。

3、配置內容通過配置管理平臺后臺服務進行持久化（保存到數據庫中）。

4、操作人員通過配置平臺進行推送操作，將配置推送到zk集群相應結點（/cfgcenter/系統名稱/應用名稱/實例名稱/分組名稱）。

5、配置中心客戶端監聽zk集群中對應結點數據發生變化，讀取變更后的內容，解析內容，刷新本地備份（分布式容災）和Spring環境變量。

6、配置中心客戶端如果和zk集群丟失連接，將加載本地本分配置到Spring環境變量。

7、配置中心客戶端重新和zk集群建立連接，從zk集群中拉取最新配置內容，解析配置，刷新本地備份（分布式容災）和Spring環境變量。

8、配置中心客戶端將Spring環境變量刷新之后，動態刷新依賴配置中心配置的bean。

配置中心代碼視圖　　

　　

配置中心客戶端設計解析

配置中心客戶端初始化

@Component public class CfgcenterInit implements ApplicationContextInitializer<ConfigurableWebApplicationContext>, ApplicationListener<ApplicationEvent> {private static Logger LOGGER = LoggerFactory.getLogger(CfgcenterInit.class);@Overridepublic void onApplicationEvent(ApplicationEvent event) {if (event instanceof ContextRefreshedEvent) {LOGGER.info("初始化配置中心客戶端監聽器...");ZKClient.getInstance().init();} else if (event instanceof RefreshEvent) {ZKClient.getInstance().getAeb().post(event);} else if (event instanceof ContextClosedEvent) {if (null != ZKClient.getInstance().getCw()) {ZKClient.getInstance().getCw().close();}}}@Overridepublic void initialize(ConfigurableWebApplicationContext cac) {try {ZookeeperProperties zookeeperProperties = ConfigurationBinder.withPropertySources(cac.getEnvironment()).bind(ZookeeperProperties.class);if (!zookeeperProperties.isEnabled()) {LOGGER.info("未開啟配置中心客戶端...");return;}ZKClient.getInstance().binding(zookeeperProperties, new ZookeeperConfigProperties(), cac);} catch (Exception e) {LOGGER.error("配置中心客戶端初始化異常...", e);}} }

1、ApplicationContextInitializer#initialize方法中，獲取zk連接信息配置，如果開啟配置中心客戶端，將ZookeeperProperties（zk集群連接信息）、ZookeeperConfigProperties（客戶端監聽zk集群結點信息）、ConfigurableWebApplicationContext （應用上下文）綁定到ZKClient實例中去。

2、ApplicationListener#onApplicationEvent方法中監聽ContextRefreshedEvent（初始化配置中心客戶端監聽器）、RefreshEvent（配置刷新事件，通過guava的事件總線進行推送）、ContextClosedEvent（關閉配置中心客戶端資源）。

配置中心客戶端監聽器

public class ConfigWatcher implements Closeable, TreeCacheListener {private static Logger LOGGER = LoggerFactory.getLogger(ConfigWatcher.class);private AtomicBoolean running = new AtomicBoolean(false);private String context;private CuratorFramework source;private HashMap<String, TreeCache> caches;public ConfigWatcher(String context, CuratorFramework source) {this.context = context;this.source = source;}public void start() {if (this.running.compareAndSet(false, true)) {this.caches = new HashMap<>();if (!context.startsWith("/")) {context = "/" + context;}try {TreeCache cache = TreeCache.newBuilder(this.source, context).build();cache.getListenable().addListener(this);cache.start();this.caches.put(context, cache);// no race condition since ZookeeperAutoConfiguration.curatorFramework// calls curator.blockUntilConnected} catch (KeeperException.NoNodeException e) {// no node, ignore} catch (Exception e) {LOGGER.error("Error initializing listener for context " + context, e);}}}@Overridepublic void close() {if (this.running.compareAndSet(true, false)) {for (TreeCache cache : this.caches.values()) {cache.close();}this.caches = null;}}@Overridepublic void childEvent(CuratorFramework client, TreeCacheEvent event) {TreeCacheEvent.Type eventType = event.getType();switch (eventType) {case INITIALIZED:LOGGER.info("配置中心客戶端同步服務端狀態完成...");refreshEnvAndBeans(event);break;case NODE_REMOVED:case NODE_UPDATED:refreshEnvAndBeans(event);break;case CONNECTION_SUSPENDED:case CONNECTION_LOST:LOGGER.info("配置中心客戶端與服務端連接異常...");break;case CONNECTION_RECONNECTED:LOGGER.info("配置中心客戶端與服務端重新建立連接...");break;}}private void refreshEnvAndBeans(TreeCacheEvent event) {//刷新環境變量 ZKClient.getInstance().refreshEnvironment();//刷新Bean ZKClient.getInstance().getAep().publishEvent(new RefreshEvent(this, event, getEventDesc(event)));}private String getEventDesc(TreeCacheEvent event) {StringBuilder out = new StringBuilder();out.append("type=").append(event.getType());TreeCacheEvent.Type eventType = event.getType();if (eventType == NODE_UPDATED|| eventType == NODE_REMOVED) {out.append(", path=").append(event.getData().getPath());byte[] data = event.getData().getData();if (data != null) {out.append(", data=").append(new String(data, StandardCharsets.UTF_8));}}return out.toString();} }

1、通過TreeCache監聽路徑/cfgcenter/系統名稱/應用名稱/實例名稱/分組名稱（該路徑下可能會存在多個子節點，每個子節點對應一份配置，每一份配置大小不能超過64k）。

2、TreeCache監聽事件類型如下

• INITIALIZED（完成同步服務端狀態，同步狀態【NODE_REMOVED、 NODE_UPDATED、CONNECTION_RECONNECTED】之后觸發）
• NODE_REMOVED（結點移除觸發）
• NODE_UPDATED（結點數據更新觸發）
• CONNECTION_SUSPENDED（連接丟失觸發）
• CONNECTION_LOST（完全丟失連接觸發）
• CONNECTION_RECONNECTED（重新連接觸發）

3、監聽到INITIALIZED、NODE_UPDATED、NODE_REMOVED事件之后，執行refreshEnvAndBeans方法，刷新spring環境變量，同時刷新spring容器相關的Bean。

?配置中心客戶端刷新spring環境變量

public class ZookeeperPropertySourceLocator {public static final String ZOOKEEPER_PREPERTY_SOURCE_NAME = "cfg-zookeeper";private ZookeeperConfigProperties properties;private CuratorFramework curator;private static Logger LOGGER = LoggerFactory.getLogger(ZookeeperPropertySourceLocator.class);public ZookeeperPropertySourceLocator(CuratorFramework curator, ZookeeperConfigProperties properties) {this.curator = curator;this.properties = properties;}public String getContext() {return this.properties.getContext();}public PropertySource getCfgcenterPropertySource(Environment environment) {ConfigurableEnvironment env = (ConfigurableEnvironment) environment;return env.getPropertySources().get(ZOOKEEPER_PREPERTY_SOURCE_NAME);}public void locate(Environment environment) {if (environment instanceof ConfigurableEnvironment) {ConfigurableEnvironment env = (ConfigurableEnvironment) environment;String context = properties.getContext();CompositePropertySource composite = new CompositePropertySource(ZOOKEEPER_PREPERTY_SOURCE_NAME);try {PropertySource propertySource = create(context);composite.addPropertySource(propertySource);if (null != env.getPropertySources().get(ZOOKEEPER_PREPERTY_SOURCE_NAME)) {LOGGER.info("替換PropertySource: " + ZOOKEEPER_PREPERTY_SOURCE_NAME);env.getPropertySources().replace(ZOOKEEPER_PREPERTY_SOURCE_NAME, composite);} else {LOGGER.info("添加PropertySource: " + ZOOKEEPER_PREPERTY_SOURCE_NAME);env.getPropertySources().addFirst(composite);}} catch (Exception e) {if (this.properties.isFailFast()) {ReflectionUtils.rethrowRuntimeException(e);} else {LOGGER.error("Unable to load zookeeper config from " + context, e);}}}}@PreDestroypublic void destroy() {}private void backupZookeeperPropertySource(ZookeeperPropertySource zps) {String backupDir = BASE_BACKUP_DIR + this.properties.getContext();String backupFile = String.format("%s/%s", backupDir, APP_NAME + ".properties");File bakFile = new File(backupFile);StringBuilder data = new StringBuilder();for (String propertyName : zps.getPropertyNames()) {data.append(propertyName).append("=").append(zps.getProperty(propertyName)).append(System.lineSeparator());}try {FileUtils.writeStringToFile(bakFile, data.toString(), Charsets.UTF_8);LOGGER.info("配置中心客戶端刷新本地備份完成, path: " + backupDir);} catch (IOException e) {LOGGER.error("配置中心客戶端刷新本地備份異常..., path: " + backupDir, e);}}private PropertySource<CuratorFramework> create(String context) {ZookeeperPropertySource zps;if (ZKClient.getInstance().isConnected()) {zps = new ZookeeperPropertySource(context, this.curator, false);this.backupZookeeperPropertySource(zps);} else {zps = new ZookeeperPropertySource(context, this.curator, true);}return zps;} }

ZookeeperPropertySourceLocator會創建ZookeeperPropertySource，然后放入Spring的Environment變量中。如果配置中心客戶端和zk集群處于連接狀態，加載完ZookeeperPropertySource之后，備份到本地。

public class ZookeeperPropertySource extends AbstractZookeeperPropertySource {private static Logger LOGGER = LoggerFactory.getLogger(ZookeeperPropertySource.class);private Map<String, String> properties = new LinkedHashMap<>();public ZookeeperPropertySource(String context, CuratorFramework source, boolean backup) {super(context, source);//加載本地配置if (backup) {String backupDir = BASE_BACKUP_DIR + this.getContext();String backupFile = String.format("%s/%s", backupDir, APP_NAME + ".properties");try {InputStream is = FileUtils.openInputStream(new File(backupFile));InputStreamReader isr = new InputStreamReader(is);Properties properties = new Properties();properties.load(isr);properties.forEach((k, v) -> this.properties.put((String) k, (String) v));} catch (Exception e) {LOGGER.error("配置中心客戶端本地配置加載異常...", e);}}//加載遠程配置else {findProperties(this.getContext(), null);}}@Overridepublic Object getProperty(String name) {return this.properties.get(name);}private byte[] getPropertyBytes(String fullPath) {try {byte[] bytes = null;try {bytes = this.getSource().getData().forPath(fullPath);} catch (KeeperException e) {if (e.code() != KeeperException.Code.NONODE) {throw e;}}return bytes;} catch (Exception exception) {ReflectionUtils.rethrowRuntimeException(exception);}return null;}@Overridepublic String[] getPropertyNames() {Set<String> strings = this.properties.keySet();return strings.toArray(new String[strings.size()]);}private void findProperties(String path, List<String> children) {try {LOGGER.info("entering findProperties for path: " + path);if (children == null) {children = getChildren(path);}if (children == null || children.isEmpty()) {return;}for (String child : children) {String childPath = path + "/" + child;List<String> childPathChildren = getChildren(childPath);byte[] bytes = getPropertyBytes(childPath);if (!ArrayUtils.isEmpty(bytes)) {registerKeyValue(childPath, new String(bytes, Charset.forName("UTF-8")));}// Check children even if we have found a value for the current znode findProperties(childPath, childPathChildren);}LOGGER.info("leaving findProperties for path: " + path);} catch (Exception exception) {ReflectionUtils.rethrowRuntimeException(exception);}}private void registerKeyValue(String path, String value) {String key = sanitizeKey(path);LOGGER.info(String.format("配置中心客戶端解析配置節點(%s)，數據:%s", key, value));try {Properties properties = new Properties();properties.load(new StringReader(value));properties.forEach((k, v) -> this.properties.put((String) k, (String) v));} catch (IOException e) {LOGGER.info(String.format("配置中心客戶端解析配置節點(%s)異常...", key));}}private List<String> getChildren(String path) throws Exception {List<String> children = null;try {children = this.getSource().getChildren().forPath(path);} catch (KeeperException e) {if (e.code() != KeeperException.Code.NONODE) {throw e;}}return children;}}

ZookeeperPropertySource通過構造參數backup來判斷是加載zk集群中的配置還是本地備份配置。

配置中心客戶端刷新Spring容器Bean

public abstract class BaseCfgcenterBean implements InitializingBean {private static Logger LOGGER = LoggerFactory.getLogger(BaseCfgcenterBean.class);@PostConstructpublic void init() {//注冊到時間總線中 ZKClient.getInstance().getAeb().register(this);}/*** z* 綁定自身目標**/protected void doBind() {Class<? extends BaseCfgcenterBean> clazz = this.getClass();if (org.springframework.util.ClassUtils.isCglibProxy(this)) {clazz = (Class<? extends BaseCfgcenterBean>) AopUtils.getTargetClass(this);}BaseCfgcenterBean target = binding(clazz, this.getDefaultResourcePath());this.copyProperties(target);}private void copyProperties(BaseCfgcenterBean target) {ReflectionUtils.doWithFields(this.getClass(), field -> {field.setAccessible(true);field.set(this, field.get(target));}, field -> AnnotatedElementUtils.isAnnotated(field, ConfigField.class));}/*** 綁定其他目標** @param clazz 目標類**/protected <T> T doBind(Class<T> clazz) {T target = binding(clazz, this.getDefaultResourcePath());if (target instanceof InitializingBean) {try {((InitializingBean) target).afterPropertiesSet();} catch (Exception e) {LOGGER.error(String.format("屬性初始化失敗[afterPropertiesSet]， class=%s", ClassUtils.getSimpleName(clazz), e));}}return target;}private <T> T binding(Class<T> clazz, String defaultResourcePath) {Optional<PropertySource> propertySource = Optional.empty();//加載配置中心配置if (ZKClient.getInstance().isZkInit()) {propertySource = Optional.ofNullable(ZKClient.getInstance().resolvePropertySource());}//加載本地配置else {Optional<ResourcePropertySource> resourcePropertySource = ResourceUtils.getResourcePropertySource(defaultResourcePath);if (resourcePropertySource.isPresent()) {propertySource = Optional.ofNullable(resourcePropertySource.get());}}if (propertySource.isPresent()) {T target;try {target = ConfigurationBinder.withPropertySources(propertySource.get()).bind(clazz);} catch (Exception e) {LOGGER.error(String.format("屬性綁定失敗， class=%s", ClassUtils.getSimpleName(clazz)), e);return null;}return target;}return null;}@Overridepublic void afterPropertiesSet() {Class<?> target = this.getClass();if (AopUtils.isAopProxy(this)) {target = AopUtils.getTargetClass(this);}LOGGER.info(String.format("%s->%s模塊引入配置中心%s...", this.getModuleName(), ClassUtils.getSimpleName(target), (ZKClient.getInstance().isConnected() ? "生效" : "無效")));}public String getModuleName() {return StringUtils.EMPTY;}@Subscribepublic void listenRefreshEvent(RefreshEvent refreshEvent) {this.afterPropertiesSet();LOGGER.info(refreshEvent.getEventDesc());this.refresh();}//通過事件進行刷新protected void refresh() {this.doBind();}//獲取本地配置默認路徑protected abstract String getDefaultResourcePath(); }

1、對象自身實現guava事件總線監聽，監聽RefreshEvent事件，觸發對象屬性刷新操作。

2、對象初始化時，注冊自身目標到guava的事件總線對象中。

3、對象屬性刷新，獲取到PropertySource對象（配置中心配置或者項目自身靜態配置），通過ConfigurationBinder工具類將配置重新綁定的對象屬性。

配置管理平臺接口

@RestController @RequestMapping("cfg") public class CfgController {private static Logger LOGGER = LoggerFactory.getLogger(CfgController.class);private static final String ZK_PATH_PATTERN0 = "/wmhcfg/projects/%s/%s";private static final String ZK_PATH_PATTERN1 = ZK_PATH_PATTERN0 + "/%s";private static final String ZK_PATH_PATTERN = ZK_PATH_PATTERN1 + "/%s";@Autowiredprivate CfgMapper mapper;@GetMapping(value = "/search", produces = MediaType.TEXT_PLAIN_VALUE)public String findCfgContents(@RequestBody @Validated SearchVO searchVO, @RequestParam(required = false) String cfgId) {List<CfgRecord> records = mapper.findRecords(searchVO);if (CollectionUtils.isEmpty(records)) {return StringUtils.EMPTY;}if (StringUtils.isNotBlank(cfgId)) {records = records.stream().filter(record -> cfgId.equals(record.getCfgId())).collect(Collectors.toList());}StringBuilder response = new StringBuilder();Properties properties = new Properties();records.forEach(record -> {try {properties.clear();properties.load(new StringReader(record.getCfgContent()));properties.forEach((key, value) -> response.append(key).append("=").append(value).append(System.lineSeparator()).append(System.lineSeparator()));} catch (IOException e) {LOGGER.error("配置解析異常...", e);}});return response.toString();}@PostMapping(value = "/send/{systemId}/{appId}/{groupId}/{cfgId}", produces = MediaType.APPLICATION_JSON_UTF8_VALUE)public BaseResponse sendCfgContent(@RequestBody String cfgContent, @PathVariable String systemId, @PathVariable String appId, @PathVariable String groupId, @PathVariable String cfgId) {BaseResponse baseResponse = new BaseResponse();baseResponse.setRestStatus(RestStatus.SUCCESS);SearchVO searchVO = new SearchVO();searchVO.setSystemId(systemId);searchVO.setAppId(appId);searchVO.setGroupId(groupId);List<CfgRecord> records = mapper.findRecords(searchVO);CfgRecord record = null;if (!CollectionUtils.isEmpty(records)) {for (CfgRecord cfgRecord : records) {if (cfgId.equals(cfgRecord.getCfgId())) {record = cfgRecord;record.setCfgContent(cfgContent);break;}}}if (null == record) {record = new CfgRecord();record.setSystemId(systemId);record.setAppId(appId);record.setGroupId(groupId);record.setCfgId(cfgId);record.setCfgContent(cfgContent);}StringBuilder cfgContentSB = new StringBuilder();Properties properties = new Properties();try {properties.load(new StringReader(record.getCfgContent()));} catch (IOException e) {LOGGER.error("配置解析異常...", e);baseResponse.setErrors(e.getMessage());baseResponse.setRestStatus(RestStatus.FAIL_50001);return baseResponse;}properties.forEach((key, value) -> cfgContentSB.append(key).append("=").append(value).append(System.lineSeparator()));record.setCfgContent(cfgContentSB.toString());if (null == record.getId()) {mapper.insertRecord(record);} else {mapper.updateRecord(record);}return baseResponse;}@PostMapping(value = "/push", produces = MediaType.APPLICATION_JSON_UTF8_VALUE)public BaseResponse pushCfgContent(@RequestBody @Validated PushVO pushVO) {BaseResponse baseResponse = new BaseResponse();baseResponse.setRestStatus(RestStatus.SUCCESS);String path = String.format(ZK_PATH_PATTERN, pushVO.getSystemId(), pushVO.getAppId(), pushVO.getGroupId(), pushVO.getCfgId());try {SearchVO searchVO = new SearchVO();searchVO.setSystemId(pushVO.getSystemId());searchVO.setAppId(pushVO.getAppId());searchVO.setGroupId(pushVO.getGroupId());List<CfgRecord> records = mapper.findRecords(searchVO);StringBuilder cfgContent = new StringBuilder();records.forEach(record -> cfgContent.append(record.getCfgContent()).append(System.lineSeparator()));if (!ZKHelper.setData(path, cfgContent.toString().getBytes())) {baseResponse.setRestStatus(RestStatus.FAIL_50001);}} catch (Exception e) {LOGGER.error("配置推送異常...", e);baseResponse.setRestStatus(RestStatus.FAIL_50001);baseResponse.setErrors(e.getMessage());}return baseResponse;}@PostMapping(value = "/create", produces = MediaType.APPLICATION_JSON_UTF8_VALUE)public BaseResponse createCfg(@RequestBody @Validated PushVO pushVO) {BaseResponse baseResponse = new BaseResponse();String path = String.format(ZK_PATH_PATTERN, pushVO.getSystemId(), pushVO.getAppId(), pushVO.getGroupId(), pushVO.getCfgId());if (ZKHelper.createPath(path)) {baseResponse.setRestStatus(RestStatus.SUCCESS);} else {baseResponse.setRestStatus(RestStatus.FAIL_50001);}return baseResponse;}@PostMapping(value = "/delete", produces = MediaType.APPLICATION_JSON_UTF8_VALUE)public BaseResponse deleteCfg(@RequestBody @Validated DeleteVO deleteVO) {BaseResponse baseResponse = new BaseResponse();String path;if (StringUtils.isBlank(deleteVO.getGroupId())) {path = String.format(ZK_PATH_PATTERN0, deleteVO.getSystemId(), deleteVO.getAppId());} else if (StringUtils.isNotBlank(deleteVO.getGroupId()) && StringUtils.isBlank(deleteVO.getCfgId())) {path = String.format(ZK_PATH_PATTERN1, deleteVO.getSystemId(), deleteVO.getAppId(), deleteVO.getGroupId());} else {path = String.format(ZK_PATH_PATTERN, deleteVO.getSystemId(), deleteVO.getAppId(), deleteVO.getGroupId(), deleteVO.getCfgId());}if (ZKHelper.deletePath(path)) {baseResponse.setRestStatus(RestStatus.SUCCESS);} else {baseResponse.setRestStatus(RestStatus.FAIL_50001);}return baseResponse;}@GetMapping(value = "/getdata", produces = MediaType.TEXT_PLAIN_VALUE)public String getData(@RequestParam String path) {return ZKHelper.getData(path);} }

　　為配置管理前端提供配置保存、配置推送、配置刪除等操作。

配置中心測試

@Component @ConfigurationProperties(prefix = "cfg.test") public class TestCfgcenterBean extends BaseCfgcenterBean {@ConfigFieldprivate String text;@ConfigFieldprivate Map<String, List<String>> map;public String getText() {return text;}public void setText(String text) {this.text = text;}public Map<String, List<String>> getMap() {return map;}public void setMap(Map<String, List<String>> map) {this.map = map;}@Overrideprotected String getDefaultResourcePath() {return StringUtils.EMPTY;}@Overrideprotected void refresh() {super.refresh();System.out.println("text=" + this.text);System.out.println("map=" + JSON.toJSONString(map));} }

TestCfgcenterBean繼承BaseCfgcenterBean，配置中心配置變更后可以自動將新的配置綁定到對象上。

@SpringBootApplication(exclude = RedissonAutoConfiguration.class) @EnableAspectJAutoProxy(proxyTargetClass = true, exposeProxy = true) @EnableRetry public class SpringbootApplication {public static void main(String[] args) {System.setProperty("xxx.system.id", "test_system");System.setProperty("xxx.app.id", "test_app");System.setProperty("groupenv", "x");SpringApplication.run(SpringbootApplication.class, args);}}

啟動類設置配置中心客戶端需要的環境變量：系統標識、項目標識、分組標識。

?客戶端與zk第一建立連接，同步完狀態之后，觸發INITIALIZED事件，刷新bean屬性配置。

客戶端與zk斷開重連之后，同步完狀態后觸發INITIALIZED事件，刷新bean屬性配置。

需要源碼

請關注訂閱號，回復：cfgcenter， 便可查看。

轉載于:https://www.cnblogs.com/hujunzheng/p/10932153.html

總結

以上是生活随笔為你收集整理的基于zookeeper实现分布式配置中心（二）的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。