闵行网站建设,网站相册代码,网站seo诊断报告例子,网站加速器前言
本文基于 Apollo 1.8.0 版本展开分析。
客户端
使用
Apollo 支持 API 方式和 Spring 整合两种方式。
API 方式
API 方式是最简单、高效使用使用 Apollo 配置的方式#xff0c;不依赖 Spring 框架即可使用。
获取命名空间的配置
// 1、获取默认的命名空间的配置
C…前言
本文基于 Apollo 1.8.0 版本展开分析。
客户端
使用
Apollo 支持 API 方式和 Spring 整合两种方式。
API 方式
API 方式是最简单、高效使用使用 Apollo 配置的方式不依赖 Spring 框架即可使用。
获取命名空间的配置
// 1、获取默认的命名空间的配置
Config config ConfigService.getAppConfig();// 2、获取properties格式的命名空间的配置
// String somePublicNamespace CAT;
// Config config ConfigService.getConfig(somePublicNamespace);// 3、获取yaml/yml格式的命名空间的配置
// Config config ConfigService.getConfig(application.yml);// 4、获取其它格式的命名空间的配置
// ConfigFile configFile ConfigService.getConfigFile(test, ConfigFileFormat.XML);
// String content configFile.getContent();String someKey someKeyFromDefaultNamespace;
String someDefaultValue someDefaultValueForTheKey;
String value config.getProperty(someKey, someDefaultValue);通过 Config 的 getProperty 方法可以获取到指定属性对应的属性值。
监听配置变化事件
Config config ConfigService.getAppConfig();
config.addChangeListener(new ConfigChangeListener() {Overridepubic void onChange(ConfigChangeEvent changeEvent) {System.out.println(Changes for namespace changeEvent.getNamespace());for (String key : changeEvent.changedKeys()) {ConfigChange change changeEvent.getChange(key);System.out.println(String.format(Found change - key: %s, oldValue: %s, newValue: %s, changeType: %s, change.getPropertyName(), change.getOldValue(), change.getNewValue(), change.getChangeType()));}}
});希望配置发生变化时得到通知。通过 Config 的 addChangeListener 方法添加一个 ConfigChangeListener 监听器。
Spring整合方式
可以在代码中直接使用如Value(${someKeyFromApollo:someDefaultValue}) 。
也可以在配置文件中使用如spring.datasource.url:${someKeyFromApollo:someDefaultValue} 。
甚至可以直接托管 Spring 中的配置。如在 Apollo 中直接配置 spring.datasource.urljdbc:mysql://localhost:3306/somedb 。
支持 Spring Boot 的 ConfigurationProperties 方式。
也支持结合 API 方式使用如ApolloConfig private Config config; 。
基于 Spring 的配置
注入默认的命名空间的配置到 Spring 中
Configuration
EnableApolloConfig
public class AppConfig {}注入多个命名空间的配置到 Spring 中
Configuration
EnableApolloConfig({FX.apollo, application.yml})
public class AnotherAppConfig {}注入多个命名空间的配置到 Spring 中并且指定顺序
在 EnableApolloConfig 注解中的 order 属性指定顺序值越小则顺序越靠前。
Configuration
EnableApolloConfig(order 2)
public class AppConfig {}Configuration
EnableApolloConfig(value {FX.apollo, application.yml}, order 1)
public class AnotherAppConfig {}基于 Spring Boot 的配置
额外支持通过 application.properties / bootstrap.properties 进行配置该方式可以使配置在更早的阶段注入比如使用 ConditionalOnProperty 的场景或者有一些 spring-boot-starter 在启动阶段就需要读取配置然后做一些事情。
# 启动阶段注入application命名空间的配置
apollo.bootstrap.enabled true也支持注入多个命名空间的配置。
# 启动阶段注入application,FX.apollo,application.yml命名空间的配置
apollo.bootstrap.enabled true
apollo.bootstrap.namespaces application,FX.apollo,application.yml可以让 Apollo 的加载顺序在日志系统之前比如希望把日志相关的配置logging.level.rootinfo 或者 logback-spring.xml 中的参数也交给 Apollo 管理。
# 启动阶段注入application命名空间的配置
apollo.bootstrap.enabled true
# 让 Apollo 的加载顺序在日志系统之前
apollo.bootstrap.eagerLoad.enabled true原理
1、客户端和服务端保持了一个长连接从而能第一时间获得配置更新的推送。
2、客户端还会定时从服务端拉取应用的最新配置。
这是一个 fallback 机制为了防止推送机制失效导致配置不更新。客户端定时拉取会上报本地版本所以一般情况下对于定时拉取的操作服务端都会返回 304 - Not Modified。定时频率默认为每 5 分钟拉取一次客户端也可以通过运行时指定 apollo.refreshInterval 系统参数来覆盖单位为分钟。
3、客户端从服务端获取到应用的最新配置后会保存在内存中。
4、客户端会把从服务端获取到的配置在本地文件系统缓存一份。在遇到服务不可用或者网络不通的时候依然能从本地恢复配置。
5、应用程序可以从 Apollo 客户端获取最新的配置、订阅配置更新通知。
源码分析
RemoteConfigRepository
在 ConfigService 的 getConfig 方法中会调用 RemoteConfigRepository 的构造器方法。
先来看 RemoteConfigRepository 构造器方法。
public RemoteConfigRepository(String namespace) {m_namespace namespace;m_configCache new AtomicReference();m_configUtil ApolloInjector.getInstance(ConfigUtil.class);m_httpUtil ApolloInjector.getInstance(HttpUtil.class);m_serviceLocator ApolloInjector.getInstance(ConfigServiceLocator.class);remoteConfigLongPollService ApolloInjector.getInstance(RemoteConfigLongPollService.class);m_longPollServiceDto new AtomicReference();m_remoteMessages new AtomicReference();m_loadConfigRateLimiter RateLimiter.create(m_configUtil.getLoadConfigQPS());m_configNeedForceRefresh new AtomicBoolean(true);m_loadConfigFailSchedulePolicy new ExponentialSchedulePolicy(m_configUtil.getOnErrorRetryInterval(),m_configUtil.getOnErrorRetryInterval() * 8);// 客户端从Apollo配置中心服务端获取到应用的最新配置后会保存在内存中同时持久化到磁盘中// 应用程序可以从客户端获取最新的配置、订阅配置更新通知this.trySync();// 客户端定时调度处理this.schedulePeriodicRefresh();// 客户端长轮询处理this.scheduleLongPollingRefresh();
}1、trySync 方法
首先看下 trySync 方法的处理逻辑。
protected boolean trySync() {try {sync();return true;} catch (Throwable ex) {Tracer.logEvent(ApolloConfigException, ExceptionUtil.getDetailMessage(ex));logger.warn(Sync config failed, will retry. Repository {}, reason: {}, this.getClass(), ExceptionUtil.getDetailMessage(ex));}return false;
}进入 sync 方法内部一窥究竟。
Override
protected synchronized void sync() {Transaction transaction Tracer.newTransaction(Apollo.ConfigService, syncRemoteConfig);try {// 从 m_configCache 缓存中获取本地配置ApolloConfig previous m_configCache.get();// 从服务端加载远程配置ApolloConfig current loadApolloConfig();// 如果本地配置与远程配置不一致即远程配置发生了变化if (previous ! current) {logger.debug(Remote Config refreshed!);// 更新 m_configCache 缓存m_configCache.set(current);// 回调所有 RepositoryChangeListener 的 onRepositoryChange 方法this.fireRepositoryChange(m_namespace, this.getConfig());}if (current ! null) {Tracer.logEvent(String.format(Apollo.Client.Configs.%s, current.getNamespaceName()),current.getReleaseKey());}transaction.setStatus(Transaction.SUCCESS);} catch (Throwable ex) {transaction.setStatus(ex);throw ex;} finally {transaction.complete();}
}不难看出 sync 方法实际上对应原理 3 - 客户端从Apollo配置中心服务端获取到应用的最新配置后会保存在内存中。
2、schedulePeriodicRefresh 方法
初始延迟 5 分钟之后每隔 5 分钟重复调度一次 trySync 方法。
private void schedulePeriodicRefresh() {logger.debug(Schedule periodic refresh with interval: {} {},m_configUtil.getRefreshInterval(), m_configUtil.getRefreshIntervalTimeUnit());// 默认初始延迟5分钟之后每隔5分钟重复调度一次m_executorService.scheduleAtFixedRate(new Runnable() {Overridepublic void run() {Tracer.logEvent(Apollo.ConfigService, String.format(periodicRefresh: %s, m_namespace));logger.debug(refresh config for namespace: {}, m_namespace);trySync();Tracer.logEvent(Apollo.Client.Version, Apollo.VERSION);}}, m_configUtil.getRefreshInterval(), m_configUtil.getRefreshInterval(),m_configUtil.getRefreshIntervalTimeUnit());
}不难看出 schedulePeriodicRefresh 方法实际上对应原理 2 - 客户端还会定时从服务端拉取应用的最新配置。
3、scheduleLongPollingRefresh 方法
客户端向服务端发起长轮询请求。实际上对应原理 1 - 客户端和服务端保持了一个长连接从而能第一时间获得配置更新的推送。
private void scheduleLongPollingRefresh() {// 交给 RemoteConfigLongPollService 处理remoteConfigLongPollService.submit(m_namespace, this);
}接着看下 RemoteConfigLongPollService 的 submit 方法是如何处理的。
public boolean submit(String namespace, RemoteConfigRepository remoteConfigRepository) {// 更新 m_longPollNamespaces 缓存boolean added m_longPollNamespaces.put(namespace, remoteConfigRepository);// 更新 m_notifications 缓存m_notifications.putIfAbsent(namespace, INIT_NOTIFICATION_ID);if (!m_longPollStarted.get()) {// 执行 startLongPolling 方法startLongPolling();}return added;
}接着看下 startLongPolling 方法的处理逻辑。
private void startLongPolling() {if (!m_longPollStarted.compareAndSet(false, true)) {return;}try {final String appId m_configUtil.getAppId();final String cluster m_configUtil.getCluster();final String dataCenter m_configUtil.getDataCenter();final String secret m_configUtil.getAccessKeySecret();// 默认 2000 毫秒final long longPollingInitialDelayInMills m_configUtil.getLongPollingInitialDelayInMills();m_longPollingService.submit(new Runnable() {Overridepublic void run() {if (longPollingInitialDelayInMills 0) {try {logger.debug(Long polling will start in {} ms., longPollingInitialDelayInMills);TimeUnit.MILLISECONDS.sleep(longPollingInitialDelayInMills);} catch (InterruptedException e) {//ignore}}// 执行 doLongPollingRefresh 方法doLongPollingRefresh(appId, cluster, dataCenter, secret);}});} catch (Throwable ex) {m_longPollStarted.set(false);ApolloConfigException exception new ApolloConfigException(Schedule long polling refresh failed, ex);Tracer.logError(exception);logger.warn(ExceptionUtil.getDetailMessage(exception));}
}接着看下 doLongPollingRefresh(…) 方法的处理逻辑。
private void doLongPollingRefresh(String appId, String cluster, String dataCenter, String secret) {final Random random new Random();ServiceDTO lastServiceDto null;while (!m_longPollingStopped.get() !Thread.currentThread().isInterrupted()) {// 限流判断if (!m_longPollRateLimiter.tryAcquire(5, TimeUnit.SECONDS)) {try {TimeUnit.SECONDS.sleep(5);} catch (InterruptedException e) {}}Transaction transaction Tracer.newTransaction(Apollo.ConfigService, pollNotification);String url null;try {if (lastServiceDto null) {// 向服务端发起一个 /services/config 的 GET 请求从响应体中得到 ServiceDTO 列表ListServiceDTO configServices getConfigServices();// 采用随机策略从 ServiceDTO 列表中选出一个 ServiceDTO 实例lastServiceDto configServices.get(random.nextInt(configServices.size()));}// 组装请求的 URLurl assembleLongPollRefreshUrl(lastServiceDto.getHomepageUrl(), appId, cluster, dataCenter,m_notifications);logger.debug(Long polling from {}, url);HttpRequest request new HttpRequest(url);// 默认读操作的超时时间为 90 秒request.setReadTimeout(LONG_POLLING_READ_TIMEOUT);if (!StringUtils.isBlank(secret)) {MapString, String headers Signature.buildHttpHeaders(url, appId, secret);request.setHeaders(headers);}transaction.addData(Url, url);// 1、底层采用 HttpURLConnection 向服务端发起一个 /notifications/v2 的 GET 请求final HttpResponseListApolloConfigNotification response m_httpUtil.doGet(request, m_responseType);logger.debug(Long polling response: {}, url: {}, response.getStatusCode(), url);if (response.getStatusCode() 200 response.getBody() ! null) {// 2、更新客户端本地 m_notifications 缓存updateNotifications(response.getBody());// 3、更新客户端本地 m_remoteNotificationMessages 缓存updateRemoteNotifications(response.getBody());transaction.addData(Result, response.getBody().toString());// 4、对比客户端本地缓存与服务端返回的配置信息如果发生变化则更新本地缓存并触发监听器回调notify(lastServiceDto, response.getBody());}if (response.getStatusCode() 304 random.nextBoolean()) {lastServiceDto null;}m_longPollFailSchedulePolicyInSecond.success();transaction.addData(StatusCode, response.getStatusCode());transaction.setStatus(Transaction.SUCCESS);} catch (Throwable ex) {lastServiceDto null;Tracer.logEvent(ApolloConfigException, ExceptionUtil.getDetailMessage(ex));transaction.setStatus(ex);long sleepTimeInSecond m_longPollFailSchedulePolicyInSecond.fail();logger.warn(Long polling failed, will retry in {} seconds. appId: {}, cluster: {}, namespaces: {}, long polling url: {}, reason: {},sleepTimeInSecond, appId, cluster, assembleNamespaces(), url, ExceptionUtil.getDetailMessage(ex));try {TimeUnit.SECONDS.sleep(sleepTimeInSecond);} catch (InterruptedException ie) {//ignore}} finally {transaction.complete();}}
}默认的监听器是 LocalFileConfigRepository。
LocalFileConfigRepository
看下 onRepositoryChange 方法它是如何处理的。
Override
public void onRepositoryChange(String namespace, Properties newProperties) {if (newProperties.equals(m_fileProperties)) {return;}Properties newFileProperties propertiesFactory.getPropertiesInstance();newFileProperties.putAll(newProperties);// 1、将配置持久化到磁盘中updateFileProperties(newFileProperties, m_upstream.getSourceType());// 2、触发监听器回调this.fireRepositoryChange(namespace, newProperties);
}1、updateFileProperties 方法
private synchronized void updateFileProperties(Properties newProperties, ConfigSourceType sourceType) {this.m_sourceType sourceType;if (newProperties.equals(m_fileProperties)) {return;}this.m_fileProperties newProperties;persistLocalCacheFile(m_baseDir, m_namespace);
}进入 persistLocalCacheFile 方法一窥究竟。
void persistLocalCacheFile(File baseDir, String namespace) {if (baseDir null) {return;}// 文件名为 ${appId} ${cluster} ${namespace}.propertiesFile file assembleLocalCacheFile(baseDir, namespace);OutputStream out null;Transaction transaction Tracer.newTransaction(Apollo.ConfigService, persistLocalConfigFile);transaction.addData(LocalConfigFile, file.getAbsolutePath());try {out new FileOutputStream(file);// 底层调用 Properties 的 store 方法将配置持久化到磁盘中m_fileProperties.store(out, Persisted by DefaultConfig);transaction.setStatus(Transaction.SUCCESS);} catch (IOException ex) {ApolloConfigException exception new ApolloConfigException(String.format(Persist local cache file %s failed, file.getAbsolutePath()), ex);Tracer.logError(exception);transaction.setStatus(exception);logger.warn(Persist local cache file {} failed, reason: {}., file.getAbsolutePath(),ExceptionUtil.getDetailMessage(ex));} finally {if (out ! null) {try {out.close();} catch (IOException ex) {//ignore}}transaction.complete();}
}简单看下 Properties 的 store 方法是如何持久化的。
public void store(OutputStream out, String comments)throws IOException
{store0(new BufferedWriter(new OutputStreamWriter(out, 8859_1)),comments,true);
}private void store0(BufferedWriter bw, String comments, boolean escUnicode)throws IOException
{if (comments ! null) {writeComments(bw, comments);}bw.write(# new Date().toString());bw.newLine();synchronized (this) {for (Enumeration? e keys(); e.hasMoreElements();) {String key (String)e.nextElement();String val (String)get(key);key saveConvert(key, true, escUnicode);val saveConvert(val, false, escUnicode);bw.write(key val);bw.newLine();}}bw.flush();
}可以看出底层使用 BufferedWriter 的 write 方法将数据写入到磁盘中。
而本地磁盘文件的路径在 findLocalCacheDir 方法中进行定义。
LocalFileConfigRepository 构造器方法中会调用该方法。
private File findLocalCacheDir() {try {String defaultCacheDir m_configUtil.getDefaultLocalCacheDir();Path path Paths.get(defaultCacheDir);if (!Files.exists(path)) {Files.createDirectories(path);}if (Files.exists(path) Files.isWritable(path)) {// 一种是通过参数指定的文件路径return new File(defaultCacheDir, CONFIG_DIR);}} catch (Throwable ex) {//ignore}// 另一种是应用程序的上下文路径return new File(ClassLoaderUtil.getClassPath(), CONFIG_DIR);
}简单看下第一种通过参数指定的文件路径。
public String getDefaultLocalCacheDir() {String cacheRoot getCustomizedCacheRoot();if (!Strings.isNullOrEmpty(cacheRoot)) {return cacheRoot File.separator getAppId();}cacheRoot isOSWindows() ? C:\\opt\\data\\%s : /opt/data/%s;return String.format(cacheRoot, getAppId());
}进入 getCustomizedCacheRoot 方法一窥究竟。
private String getCustomizedCacheRoot() {// 1. Get from System PropertyString cacheRoot System.getProperty(apollo.cacheDir);if (Strings.isNullOrEmpty(cacheRoot)) {// 2. Get from OS environment variablecacheRoot System.getenv(APOLLO_CACHEDIR);}if (Strings.isNullOrEmpty(cacheRoot)) {// 3. Get from server.propertiescacheRoot Foundation.server().getProperty(apollo.cacheDir, null);}if (Strings.isNullOrEmpty(cacheRoot)) {// 4. Get from app.propertiescacheRoot Foundation.app().getProperty(apollo.cacheDir, null);}return cacheRoot;
}2、fireRepositoryChange 方法
实际上对应原理 5 - 应用程序可以从 Apollo 客户端获取最新的配置、订阅配置更新通知。
protected void fireRepositoryChange(String namespace, Properties newProperties) {for (RepositoryChangeListener listener : m_listeners) {try {listener.onRepositoryChange(namespace, newProperties);} catch (Throwable ex) {Tracer.logError(ex);logger.error(Failed to invoke repository change listener {}, listener.getClass(), ex);}}
}默认的监听器是 DefaultConfig。
DefaultConfig
看下 DefaultConfig 是如何回调处理的。
Override
public synchronized void onRepositoryChange(String namespace, Properties newProperties) {if (newProperties.equals(m_configProperties.get())) {return;}ConfigSourceType sourceType m_configRepository.getSourceType();Properties newConfigProperties propertiesFactory.getPropertiesInstance();newConfigProperties.putAll(newProperties);MapString, ConfigChange actualChanges updateAndCalcConfigChanges(newConfigProperties, sourceType);if (actualChanges.isEmpty()) {return;}// 监听器回调处理this.fireConfigChange(new ConfigChangeEvent(m_namespace, actualChanges));Tracer.logEvent(Apollo.Client.ConfigChanges, m_namespace);
}进入 fireConfigChange 方法一窥究竟。
protected void fireConfigChange(final ConfigChangeEvent changeEvent) {for (final ConfigChangeListener listener : m_listeners) {if (!isConfigChangeListenerInterested(listener, changeEvent)) {continue;}m_executorService.submit(new Runnable() {Overridepublic void run() {String listenerName listener.getClass().getName();Transaction transaction Tracer.newTransaction(Apollo.ConfigChangeListener, listenerName);try {// 触发监听器回调listener.onChange(changeEvent);transaction.setStatus(Transaction.SUCCESS);} catch (Throwable ex) {transaction.setStatus(ex);Tracer.logError(ex);logger.error(Failed to invoke config change listener {}, listenerName, ex);} finally {transaction.complete();}}});}监听器默认是 AutoUpdateConfigChangeListener。
AutoUpdateConfigChangeListener
看下 AutoUpdateConfigChangeListener 是如何回调处理的。
Override
public void onChange(ConfigChangeEvent changeEvent) {// 获取发生变化的属性集合SetString keys changeEvent.changedKeys();if (CollectionUtils.isEmpty(keys)) {return;}for (String key : keys) {CollectionSpringValue targetValues springValueRegistry.get(beanFactory, key);if (targetValues null || targetValues.isEmpty()) {continue;}// 更新应用程序使用到的属性对应的属性值for (SpringValue val : targetValues) {updateSpringValue(val);}}
}进入 updateSpringValue 方法一窥究竟。
private void updateSpringValue(SpringValue springValue) {try {// 获取经过解析后的属性值Object value resolvePropertyValue(springValue);// 更新应用程序使用到的属性对应的属性值springValue.update(value);// 日志打印logger.info(Auto update apollo changed value successfully, new value: {}, {}, value,springValue);} catch (Throwable ex) {logger.error(Auto update apollo changed value failed, {}, springValue.toString(), ex);}
}SpringValue
简单看下如何更新应用程序使用到的属性对应的属性值的。
public void update(Object newVal) throws IllegalAccessException, InvocationTargetException {if (isField()) {injectField(newVal);} else {injectMethod(newVal);}
}private void injectField(Object newVal) throws IllegalAccessException {Object bean beanRef.get();if (bean null) {return;}boolean accessible field.isAccessible();field.setAccessible(true);// 底层使用 Field 的 set 方法更新属性值field.set(bean, newVal);field.setAccessible(accessible);
}private void injectMethod(Object newVal)throws InvocationTargetException, IllegalAccessException {Object bean beanRef.get();if (bean null) {return;}// 底层使用 Method 的 invoke 方法更新属性值methodParameter.getMethod().invoke(bean, newVal);
}
