聊聊Spring Cloud Config的ConfigClientWatch

序

Spring Cloud Config提供了一个ConfigClientWatch功能，可以定时轮询客户端配置的状态，如果状态发生变化，则refresh。

配置文件

spring:
cloud:
config:
uri: http://localhost:8888 watch:
enabled: true
initialDelay: 5000 ##default 180000 ms
delay: 10000 ##default 500 ms

配置类

spring-cloud-config-client-1.3.1.RELEASE-sources.jar!/org/springframework/cloud/config/client/ConfigClientAutoConfiguration.java

@Configuration
@ConditionalOnClass(ContextRefresher.class)
@ConditionalOnBean(ContextRefresher.class)
@ConditionalOnProperty(value = "spring.cloud.config.watch.enabled")
protected static class ConfigClientWatchConfiguration {

@Bean
public ConfigClientWatch configClientWatch(ContextRefresher contextRefresher) {
return new ConfigClientWatch(contextRefresher);
}
}

ConfigClientWatch

public class ConfigClientWatch implements Closeable, EnvironmentAware {

private static Log log = LogFactory
.getLog(ConfigServicePropertySourceLocator.class);

private final AtomicBoolean running = new AtomicBoolean(false);
private final ContextRefresher refresher;
private Environment environment;

public ConfigClientWatch(ContextRefresher refresher) {
this.refresher = refresher;
}

@Override
public void setEnvironment(Environment environment) {
this.environment = environment;
}

@PostConstruct
public void start() {
this.running.compareAndSet(false, true);
}

@Scheduled(initialDelayString = "${spring.cloud.config.watch.initialDelay:180000}", fixedDelayString = "${spring.cloud.config.watch.delay:500}")
public void watchConfigServer() {
if (this.running.get()) {
String newState = this.environment.getProperty("config.client.state");
String oldState = ConfigClientStateHolder.getState();

// only refresh if state has changed
if (stateChanged(oldState, newState)) {
ConfigClientStateHolder.setState(newState);
this.refresher.refresh();
}
}
}

/* for testing */ boolean stateChanged(String oldState, String newState) {
return (!hasText(oldState) && hasText(newState))
|| (hasText(oldState) && !oldState.equals(newState));
}

@Override
public void close() {
this.running.compareAndSet(true, false);
}

}

依赖config.client.state的环境变量，来判断client端配置文件的状态
依赖ContextRefresher去刷新配置/实例

ContextRefresher

spring-cloud-context-1.2.2.RELEASE-sources.jar!/org/springframework/cloud/context/refresh/ContextRefresher.java

public synchronized Set<String> refresh() {
Map<String, Object> before = extract(
this.context.getEnvironment().getPropertySources());
addConfigFilesToEnvironment();
Set<String> keys = changes(before,
extract(this.context.getEnvironment().getPropertySources())).keySet();
this.context.publishEvent(new EnvironmentChangeEvent(keys));
this.scope.refreshAll();
return keys;
}

这个refresh主要做两件事情:

第一件是发布EnvironmentChangeEvent事件

第二件是调用RefreshScope的refreshAll方法

EnvironmentChangeEvent的listener

spring-cloud-context-1.2.2.RELEASE-sources.jar!/org/springframework/cloud/context/properties/ConfigurationPropertiesRebinder.java

@Component
@ManagedResource
public class ConfigurationPropertiesRebinder
implements ApplicationContextAware, ApplicationListener<EnvironmentChangeEvent> {

private ConfigurationPropertiesBeans beans;

private ConfigurationPropertiesBindingPostProcess
7fe0
or binder;

private ApplicationContext applicationContext;

private Map<String, Exception> errors = new ConcurrentHashMap<>();

public ConfigurationPropertiesRebinder(
ConfigurationPropertiesBindingPostProcessor binder,
ConfigurationPropertiesBeans beans) {
this.binder = binder;
this.beans = beans;
}

@Override
public void setApplicationContext(ApplicationContext applicationContext)
throws BeansException {
this.applicationContext = applicationContext;
}

/**
* A map of bean name to errors when instantiating the bean.
*
* @return the errors accumulated since the latest destroy
*/
public Map<String, Exception> getErrors() {
return this.errors;
}

@ManagedOperation
public void rebind() {
this.errors.clear();
for (String name : this.beans.getBeanNames()) {
rebind(name);
}
}

@ManagedOperation
public boolean rebind(String name) {
if (!this.beans.getBeanNames().contains(name)) {
return false;
}
if (this.applicationContext != null) {
try {
Object bean = this.applicationContext.getBean(name);
if (AopUtils.isCglibProxy(bean)) {
bean = getTargetObject(bean);
}
this.binder.postProcessBeforeInitialization(bean, name);
this.applicationContext.getAutowireCapableBeanFactory()
.initializeBean(bean, name);
return true;
}
catch (RuntimeException e) {
this.errors.put(name, e);
throw e;
}
}
return false;
}

@SuppressWarnings("unchecked")
private static <T> T getTargetObject(Object candidate) {
try {
if (AopUtils.isAopProxy(candidate) && (candidate instanceof Advised)) {
return (T) ((Advised) candidate).getTargetSource().getTarget();
}
}
catch (Exception ex) {
throw new IllegalStateException("Failed to unwrap proxied object", ex);
}
return (T) candidate;
}

@ManagedAttribute
public Set<String> getBeanNames() {
return new HashSet<String>(this.beans.getBeanNames());
}

@Override
public void onApplicationEvent(EnvironmentChangeEvent event) {
rebind();
}

}

RefreshScope的refreshAll

spring-cloud-context-1.2.2.RELEASE-sources.jar!/org/springframework/cloud/context/scope/refresh/RefreshScope.java

@ManagedOperation(description = "Dispose of the current instance of all beans in this scope and force a refresh on next method execution.")
public void refreshAll() {
super.destroy();
this.context.publishEvent(new RefreshScopeRefreshedEvent());
}

发布RefreshScopeRefreshedEvent事件

RefreshScopeRefreshedEvent事件listener

spring-cloud-netflix-eureka-client-1.3.1.RELEASE-sources.jar!/org/springframework/cloud/netflix/eureka/EurekaDiscoveryClientConfiguration.java

@Configuration
@ConditionalOnClass(RefreshScopeRefreshedEvent.class)
protected static class EurekaClientConfigurationRefresher {

@Autowired(required = false)
private EurekaClient eurekaClient;

@Autowired(required = false)
private EurekaAutoServiceRegistration autoRegistration;

@EventListener(RefreshScopeRefreshedEvent.class)
public void onApplicationEvent(RefreshScopeRefreshedEvent event) {
//This will force the creation of the EurkaClient bean if not already created
//to make sure the client will be reregistered after a refresh event
if(eurekaClient != null) {
eurekaClient.getApplications();
}
if (autoRegistration != null) {
// register in case meta data changed
this.autoRegistration.stop();
this.autoRegistration.start();
}
}
}

spring-cloud-netflix-core-1.2.6.RELEASE-sources.jar!/org/springframework/cloud/netflix/zuul/ZuulConfiguration.java

@Bean
public ApplicationListener<ApplicationEvent> zuulRefreshRoutesListener() {
return new ZuulRefreshListener();
}
private static class ZuulRefreshListener
implements ApplicationListener<ApplicationEvent> {

@Autowired
private ZuulHandlerMapping zuulHandlerMapping;

private HeartbeatMonitor heartbeatMonitor = new HeartbeatMonitor();

@Override
public void onApplicationEvent(ApplicationEvent event) {
if (event instanceof ContextRefreshedEvent
|| event instanceof RefreshScopeRefreshedEvent
|| event instanceof RoutesRefreshedEvent) {
this.zuulHandlerMapping.setDirty(true);
}
else if (event instanceof HeartbeatEvent) {
if (this.heartbeatMonitor.update(((HeartbeatEvent) event).getValue())) {
this.zuulHandlerMapping.setDirty(true);
}
}
}

}

小结

Spring Cloud Config的代码是有提供ConfigClientWatch，但是实际单纯使用git作为config server的时候，拉取配置的时候得到的state始终是null，因此客户端轮询是起不到刷新效果的。第二个就是这个refresh发布的RefreshScopeRefreshedEvent，eureka会先去更新注册信息为DOWN，然后再UP起来，这个频繁操作有点风险。