Apollo源碼閱讀筆記（二）

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都會被觸發更新。