欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识！
您现在的位置是： 首页

Apollo源码阅读笔记（二）

程序员文章站 2022-03-05 18:22:01
...

Apollo源码阅读笔记（二）

前面 分析了apollo配置设置到Spring的environment的过程，此文继续PropertySourcesProcessor.postProcessBeanFactory里面调用的第二个方法initializeAutoUpdatePropertiesFeature(beanFactory)，其实也就是配置修改后更新相关处理逻辑。

在继续分析之前，先来看下配置是怎么自动更新的。

1. 配置更新简单示例

通过portal页面，修改配置之后，我们可以通过@ApolloConfigChangeListener来自己监听配置的变化手动更新，也可以通过@Value标记字段，字段值会自动更新。

1.1 更新event相关对象

无论是哪种方式，都离不开event，所以先了解下event相关的对象

// change事件
public class ConfigChangeEvent {
  private final String m_namespace;
  private final Map<String, ConfigChange> m_changes;
  // 省略了全部方法  
}

// change实体bean
public class ConfigChange {
  private final String namespace;
  private final String propertyName;
  private String oldValue;
  private String newValue;
  private PropertyChangeType changeType;
  // 省略了全部方法  
}

// change类型
public enum PropertyChangeType {
  ADDED, MODIFIED, DELETED
}

1.2 Value注解方式

直接在属性字段上标记@Value就可以了。

eg：

@Service
public class XXXService {
    @Value("${config.key}")
    private String XXConfig;
    ....
}

如果修改了config.key配置项，那么这里的XXConfig的值是会自动更新的。

1.3 自定义ConfigChangeListener方式

@Value方式实现起来很简单，但如果要更灵活点，比如加上一些自己的业务处理，那就需要用到@ApolloConfigChangeListener了。

@ApolloConfigChangeListener
private void onChange(ConfigChangeEvent changeEvent) {
    logger.info("配置参数发生变化[{}]", JSON.toJSONString(changeEvent));
    doSomething();
}

标记有@ApolloConfigChangeListener的这个方法，必须带一个ConfigChangeEvent的入参，通过这个event可以拿到事件的类型、变化的具体字段、变化前后值。

拿到event之后，我们可以根据具体的变化做不同业务处理。

以上是更新的使用，下面来深入研究下源码的实现。

2. 配置更新监听listener

先继续文章开头留下的尾巴 initializeAutoUpdatePropertiesFeature方法。

2.1 AutoUpdateConfigChangeListener

PropertySourcesProcessor.postProcessBeanFactory –> initializeAutoUpdatePropertiesFeature中。具体逻辑如下：

  1. 构造一个 AutoUpdateConfigChangeListener 对象 [implements ConfigChangeListener]；

  2. 拿到前面处理的所有的ConfigPropertySource组成的list，遍历ConfigPropertySource，设置listener

    ​ configPropertySource.addChangeListener(autoUpdateConfigChangeListener);

    ​ 这里加事件，最终是加在Config上了。

我们前面使用的@Value注解标记的字段，在字段值发生变化时，就是通过这里加的listener，收到通知的。

@Value是org.springframework.beans.factory.annotation.value，Spring的注解。apollo通过自己的SpringValueProcessor来处理它。来看下完整的流程：

  1. SpringValueProcessor继承了ApolloProcessor，间接实现了BeanPostProcessor

  2. 在启动的时候，会被调到 postProcessBeforeInitialization，进而调到processField
    1. 判断字段是否被@Value标记，如果没有则返回，有则继续往下
    2. 解析@Value注解里面设置的value
    3. 构造SpringValue对象 new SpringValue(key, value.value(), bean, beanName, field, false)
    4. 将构造的springValue存到SpringValueRegistry的Map<BeanFactory, Multimap<String, SpringValue>>中

  3. 当修改配置发布事件后，AutoUpdateConfigChangeListener就被触发onChange(ConfigChangeEvent changeEvent)事件

    1. 通过event拿到所有变更的keys

    2. 遍历keys，通过springValueRegistry.get(beanFactory, key) 拿到SpringValue集合对象

      ​ 这里就是从前面的2.4的map里面获取的

    3. 判断配置是否真的发生了变化 shouldTriggerAutoUpdate(changeEvent, key)

    4. 遍历SpringValue集合，逐一通过反射改变字段的值

到这里，@Value的更新流程就清楚了。

下面来看看自定义listener是怎么通知更新的。

2.2 ConfigChangeListener

更多的时候，我们是通过在方法上标记@ApolloConfigChangeListener来实现自己的监听处理。[例子见1.3代码]

通过@ApolloConfigChangeListener注解添加的监听方法，默认关注的application namespace下的全部配置项。

有关该注解的处理逻辑在 com.ctrip.framework.apollo.spring.annotation.ApolloAnnotationProcessor 中，我们重点关注如下代码段 processMethod ：

protected void processMethod(final Object bean, String beanName, final Method method) {
  ApolloConfigChangeListener annotation = AnnotationUtils.findAnnotation(method, ApolloConfigChangeListener.class);
  if (annotation == null) {
    return;
  }
  Class<?>[] parameterTypes = method.getParameterTypes();
  Preconditions.checkArgument(parameterTypes.length == 1, "Invalid number of parameters: %s for method: %s, should be 1", parameterTypes.length, method);
  Preconditions.checkArgument(ConfigChangeEvent.class.isAssignableFrom(parameterTypes[0]), "Invalid parameter type: %s for method: %s, should be ConfigChangeEvent", parameterTypes[0], method);

  ReflectionUtils.makeAccessible(method);
  String[] namespaces = annotation.value();
  String[] annotatedInterestedKeys = annotation.interestedKeys();
  Set<String> interestedKeys = annotatedInterestedKeys.length > 0 ? Sets.newHashSet(annotatedInterestedKeys) : null;
  // 创建listener  
  ConfigChangeListener configChangeListener = new ConfigChangeListener() {
    @Override
    public void onChange(ConfigChangeEvent changeEvent) {
      ReflectionUtils.invokeMethod(method, bean, changeEvent);
    }
  };

  // 给config设置listener
  for (String namespace : namespaces) {
    Config config = ConfigService.getConfig(namespace);
    if (interestedKeys == null) {
      config.addChangeListener(configChangeListener);
    } else {
      config.addChangeListener(configChangeListener, interestedKeys);
    }
  }
}

经过这段代码处理，如果有change事件，我们通过@ApolloConfigChangeListener自定义的listener就会收到消息了。

前面了解完了监听，下面来看下事件的发布。

3. 配置更新事件发布

后台portal页面修改发布之后，client端怎么接收到事件呢？其实在client启动后，就会和服务端建一个长连接。代码见 com.ctrip.framework.apollo.internals.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);
    gson = new Gson();
    this.trySync();
    this.schedulePeriodicRefresh();
    this.scheduleLongPollingRefresh();
  }

从上面构造方法最后几行可以看出，启动的时候，会先尝试从server端同步，然后会启动2个定时刷新任务，一个是定时刷新，一个是长轮询。不过不管是哪种方式，最终进入的还是 trySync() 方法。

以com.ctrip.framework.apollo.internals.RemoteConfigLongPollService为例

RemoteConfigLongPollService

—> startLongPolling()

—> doLongPollingRefresh(appId, cluster, dataCenter)，发起http长连接

—> 收到http response（server端发布了新的配置）

—> notify(lastServiceDto, response.getBody());

—> remoteConfigRepository.onLongPollNotified(lastServiceDto, remoteMessages);

—> 异步调用 remoteConfigRepository.trySync();

在trySync方法中，直接调用sync()后就返回，所以需要看sync()方法内部逻辑。在sync()方法内：

  1. 先获取本机缓存的当前配置 ApolloConfig previous = m_configCache.get()

  2. 获取server端的最新配置 ApolloConfig current = loadApolloConfig()

  3. 如果 previous != current ，更新m_configCache

    ​ m_configCache.set(current);
    ​ this.fireRepositoryChange(m_namespace, this.getConfig());

  4. 在fireRepositoryChange里面，遍历当前namespace下的listener，调用RepositoryChangeListener事件

    ​ listener.onRepositoryChange(namespace, newProperties)

    ​ <—这里事件就传到LocalFileConfigRepository 这一层了

  5. 本机该namespace的LocalFileConfigRepository实现了RepositoryChangeListener，所以会受到通知调用

    在LocalFileConfigRepository的onRepositoryChange方法中：

    1. 比较newProperties.equals(m_fileProperties)，相同就直接return，否则继续往下

    2. 更新本机缓存文件 updateFileProperties

    3. 触发 fireRepositoryChange(namespace, newProperties); <—这里事件就传到Config这一层了

    4. 触发事件在 DefaultConfig 中得到响应处理。这里先对发生变化的配置做一些处理，然后发ConfigChangeEvent事件

      ​ this.fireConfigChange(new ConfigChangeEvent(m_namespace, actualChanges));

    5. 在DefaultConfig的this.fireConfigChange里面，就会遍历 listeners，依次调用onChange方法

      ​ 准确的说，是在父类AbstractConfig中实现的；

      ​ 这里的listeners就有前面提到的AutoUpdateConfigChangeListener 和 ApolloAnnotationProcessor 中定义的ConfigChangeListener

至此，事件的发布监听就形成闭环了，这里fireConfigChange(ConfigChangeEvent)后，@Value标记的字段、@ApolloConfigChangeListener都会被触发更新。

上一篇： postman--断言写法

下一篇： postman-编写脚本

推荐阅读