上一篇(基於zookeeper實現分佈式配置中心(一))講述了zookeeper相關概念和工做原理。接下來根據zookeeper的特性,簡單實現一個分佈式配置中心。html
一、各環境配置集中管理。前端
二、配置更改,實時推送,jvm環境變量及時生效。node
三、依靠配置變動,動態擴展功能,減小二次上線帶來的成本。spring
四、減小開發人員、運維人員修改配置帶來的額外開銷。數據庫
一、配置管理平臺中,操做人員能夠建立項目所屬系統、應用名稱、實例名稱、配置分組等信息。架構
二、配置管理平臺中,操做人員能夠上傳配置文件,對屬性有增、刪、改、查的操做。app
三、配置內容經過配置管理平臺後臺服務進行持久化(保存到數據庫中)。運維
四、操做人員經過配置平臺進行推送操做,將配置推送到zk集羣相應結點(/cfgcenter/系統名稱/應用名稱/實例名稱/分組名稱)。jvm
五、配置中心客戶端監聽zk集羣中對應結點數據發生變化,讀取變動後的內容,解析內容,刷新本地備份(分佈式容災)和Spring環境變量。分佈式
六、配置中心客戶端若是和zk集羣丟失鏈接,將加載本地本分配置到Spring環境變量。
七、配置中心客戶端從新和zk集羣創建鏈接,從zk集羣中拉取最新配置內容,解析配置,刷新本地備份(分佈式容災)和Spring環境變量。
八、配置中心客戶端將Spring環境變量刷新以後,動態刷新依賴配置中心配置的bean。
@Component public class CfgcenterInit implements ApplicationContextInitializer<ConfigurableWebApplicationContext>, ApplicationListener<ApplicationEvent> { private static Logger LOGGER = LoggerFactory.getLogger(CfgcenterInit.class); @Override public 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(); } } } @Override public 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); } } }
一、ApplicationContextInitializer#initialize方法中,獲取zk鏈接信息配置,若是開啓配置中心客戶端,將ZookeeperProperties(zk集羣鏈接信息)、ZookeeperConfigProperties(客戶端監聽zk集羣結點信息)、ConfigurableWebApplicationContext (應用上下文)綁定到ZKClient實例中去。
二、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); } } } @Override public void close() { if (this.running.compareAndSet(true, false)) { for (TreeCache cache : this.caches.values()) { cache.close(); } this.caches = null; } } @Override public 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(); } }
一、經過TreeCache監聽路徑/cfgcenter/系統名稱/應用名稱/實例名稱/分組名稱(該路徑下可能會存在多個子節點,每一個子節點對應一份配置,每一份配置大小不能超過64k)。
二、TreeCache監聽事件類型以下
三、監聽到INITIALIZED、NODE_UPDATED、NODE_REMOVED事件以後,執行refreshEnvAndBeans方法,刷新spring環境變量,同時刷新spring容器相關的Bean。
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); } } } } @PreDestroy public 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); } } @Override public 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; } @Override public 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集羣中的配置仍是本地備份配置。
public abstract class BaseCfgcenterBean implements InitializingBean { private static Logger LOGGER = LoggerFactory.getLogger(BaseCfgcenterBean.class); @PostConstruct public 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; } @Override public 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; } @Subscribe public void listenRefreshEvent(RefreshEvent refreshEvent) { this.afterPropertiesSet(); LOGGER.info(refreshEvent.getEventDesc()); this.refresh(); } //經過事件進行刷新 protected void refresh() { this.doBind(); } //獲取本地配置默認路徑 protected abstract String getDefaultResourcePath(); }
一、對象自身實現guava事件總線監聽,監聽RefreshEvent事件,觸發對象屬性刷新操做。
二、對象初始化時,註冊自身目標到guava的事件總線對象中。
三、對象屬性刷新,獲取到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"; @Autowired private 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 { @ConfigField private String text; @ConfigField private 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; } @Override protected String getDefaultResourcePath() { return StringUtils.EMPTY; } @Override protected 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, 即可查看。