为什么apollo客户端可以实时获取配置

女爷i 2023-06-30 12:58 1阅读 0赞

> 普通情况下，使用 实现apollo的监听器ConfigChangeListener，通过onChange方法来实时获取配置。
> 
> 但是，如果对配置变化的时效性要求不高，只是想在用到配置的时候，使用新配置的话，可以不实现监听器，直接从environment里面获取配置。

### 文章目录 ###

*   *   *  apollo机制
         *   *  图解
             *  代码分析
             *  主要流程
             *   *  创建RemoteConfigRepository
                 *  首次同步apollo
                 *  开启定时刷新配置
                 *  开启长轮询刷新配置
                 *  fireRepositoryChange

### apollo机制 ###

#### 图解 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjI2MDI3MA_size_16_color_FFFFFF_t_70]

#### 代码分析 ####

还是通过老朋友refresh方法来层层揭晓吧。

首先，通过AbstractApplicationContext.refresh()中：

invokeBeanFactoryPostProcessors(beanFactory);

经历重重磨难，找到了PropertySourcesProcessor,并执行其postProcessBeanFactory方法，进入initializePropertySources()。

#### 主要流程 ####

1.创建Config

2.用CompositePropertySource包装Config，并塞入environment

private void initializePropertySources() {
      if (environment.getPropertySources().contains(PropertySourcesConstants.APOLLO_PROPERTY_SOURCE_NAME)) {
        // 已经初始化过apollo配置就直接返回
        return;
      }
        // 创建一个PropertySource，用于装入所有的配置
      CompositePropertySource composite = new CompositePropertySource(PropertySourcesConstants.APOLLO_PROPERTY_SOURCE_NAME);
    ......
    
      while (iterator.hasNext()) {
        int order = iterator.next();
        for (String namespace : NAMESPACE_NAMES.get(order)) {
            // 获取一个config,如果没有，则会创建一个config（后续分析，很重要哦）
          Config config = ConfigService.getConfig(namespace);
            // 用CompositePropertySource包装config为一个PropertySource，后续加入值environment
          composite.addPropertySource(configPropertySourceFactory.getConfigPropertySource(namespace, config));
        }
      }
    
      // clean up
      NAMESPACE_NAMES.clear();
    
      // 如果存在ApolloBootstrapPropertySources，则要保证ApolloBootstrapPropertySources在第一个，并且将上述的composite设置在其后，否则就是将composite设置在第一个
      if (environment.getPropertySources()
          .contains(PropertySourcesConstants.APOLLO_BOOTSTRAP_PROPERTY_SOURCE_NAME)) {
        ensureBootstrapPropertyPrecedence(environment);
        environment.getPropertySources()
            .addAfter(PropertySourcesConstants.APOLLO_BOOTSTRAP_PROPERTY_SOURCE_NAME, composite);
      } else {
        // 将composite设置在第一个后，即便我们不适用listener方式，被通知配置的变更，也能直接从environment里面通过getProperty的方式直接获取配置
        environment.getPropertySources().addFirst(composite);
      }
    }

从上，我们可以知道，最终的配置其实是放入environment里面了。故而，我们可以通过environment直接获取配置，只是不如通过监听器的onChange方法获取配置实时。

那么，了解完两部曲，就来细细品品其中的机制吧。一个config，为啥就能让你实时知道apollo的变更呢？

##### 创建RemoteConfigRepository #####

之前说的 ConfigService.getConfig(namespace) ，获取不到config，就会创建一个config的呀。此时创建的就是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();
      // 首次同步apollo
      this.trySync();
      // 定时刷新配置（大部分情况返回304，定时刷新在于可以防止长轮询失败）
      this.schedulePeriodicRefresh();
      // 长轮询刷新配置（最主要的实时获取配置的途径）
      this.scheduleLongPollingRefresh();
    }

##### 首次同步apollo #####

@Override
    protected synchronized void sync() {
      Transaction transaction = Tracer.newTransaction("Apollo.ConfigService", "syncRemoteConfig");
    
      try {
        ApolloConfig previous = m_configCache.get();
        // 根据你设定的apollo地址，appId，maxRetries等信息，发送get请求，获取当前apollo配置信息
        ApolloConfig current = loadApolloConfig();
    
        // 更新本地的apollo配置信息
        if (previous != current) {
          logger.debug("Remote Config refreshed!");
          m_configCache.set(current);
          // 刷新配置的任务可能会调用该方法，获取配置，并且通知客户端的监听器
          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();
      }
    }

##### 开启定时刷新配置 #####

private void schedulePeriodicRefresh() {
      logger.debug("Schedule periodic refresh with interval: {} {}",
          m_configUtil.getRefreshInterval(), m_configUtil.getRefreshIntervalTimeUnit());
      m_executorService.scheduleAtFixedRate(
          new Runnable() {
            @Override
            public void run() {
              Tracer.logEvent("Apollo.ConfigService", String.format("periodicRefresh: %s", m_namespace));
              logger.debug("refresh config for namespace: {}", m_namespace);
              // 和上述首次同步的trySync是同一个方法
              trySync();
              Tracer.logEvent("Apollo.Client.Version", Apollo.VERSION);
            }
              //时间间隔默认是5分钟
          }, m_configUtil.getRefreshInterval(), m_configUtil.getRefreshInterval(),
          m_configUtil.getRefreshIntervalTimeUnit());
    }

##### 开启长轮询刷新配置 #####

public boolean submit(String namespace, RemoteConfigRepository remoteConfigRepository) {
      boolean added = m_longPollNamespaces.put(namespace, remoteConfigRepository);
      m_notifications.putIfAbsent(namespace, INIT_NOTIFICATION_ID);
      if (!m_longPollStarted.get()) {
        // 开启拉取  
        startLongPolling();
      }
      return added;
    }
    
    private void doLongPollingRefresh(String appId, String cluster, String dataCenter) {
      final Random random = new Random();
      ServiceDTO lastServiceDto = null;
      while (!m_longPollingStopped.get() && !Thread.currentThread().isInterrupted()) {
        if (!m_longPollRateLimiter.tryAcquire(5, TimeUnit.SECONDS)) {
          // 等待5秒
          try {
            TimeUnit.SECONDS.sleep(5);
          } catch (InterruptedException e) {
          }
        }
          
          // 获取各种信息，发起请求，看apollo中配置有无变更
          ......
              
          // 配置发生变化，则主动获取新配置，且调用ConfigChangeListener的实现类的onChange方法    
          if (response.getStatusCode() == 200 && response.getBody() != null) {
            updateNotifications(response.getBody());
            updateRemoteNotifications(response.getBody());
            transaction.addData("Result", response.getBody().toString());
            notify(lastServiceDto, response.getBody());
          }
           ......
        } finally {
          transaction.complete();
        }
      }
    }

notify方法会最终进入fireRepositoryChange

##### fireRepositoryChange #####

该方法是之前介绍的sync()方法里面提到的，两类刷新配置的任务都会执行此方法，来通知各个监听器，有哪些配置变更。  
1.获取新配置

2.统计各种配置的变更

3.通知各个监听器

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);
        }
      }
    }
    
    @Override
    public synchronized void onRepositoryChange(String namespace, Properties newProperties) {
      if (newProperties.equals(m_configProperties.get())) {
        return;
      }
    
      ConfigSourceType sourceType = m_configRepository.getSourceType();
      Properties newConfigProperties = new Properties();
      newConfigProperties.putAll(newProperties);
    
      // 整理配置信息，确定配置的变更类型
      Map<String, ConfigChange> actualChanges = updateAndCalcConfigChanges(newConfigProperties, sourceType);
    
      //check double checked result
      if (actualChanges.isEmpty()) {
        return;
      }
    
      // 调用各个ConfigChangeListener实现类的onChange方法，发送配置信息
      this.fireConfigChange(new ConfigChangeEvent(m_namespace, actualChanges));
    
      Tracer.logEvent("Apollo.Client.ConfigChanges", m_namespace);
    }
    
    private Map<String, ConfigChange> updateAndCalcConfigChanges(Properties newConfigProperties,
        ConfigSourceType sourceType) {
      // 初次统计变化的配置  
      List<ConfigChange> configChanges =
          calcPropertyChanges(m_namespace, m_configProperties.get(), newConfigProperties);
    
      ImmutableMap.Builder<String, ConfigChange> actualChanges =
          new ImmutableMap.Builder<>();
    
      /** === Double check since DefaultConfig has multiple config sources ==== **/
    
      //1. 为配置设置旧值
      for (ConfigChange change : configChanges) {
        change.setOldValue(this.getProperty(change.getPropertyName(), change.getOldValue()));
      }
    
      //2. 更新 m_configProperties
      updateConfig(newConfigProperties, sourceType);
      clearConfigCache();
    
      //3. 遍历所有的新配置，最后确认各个配置的type（ADDED/MODIFIED/DELETED）
      for (ConfigChange change : configChanges) {
        change.setNewValue(this.getProperty(change.getPropertyName(), change.getNewValue()));
        switch (change.getChangeType()) {
          case ADDED:
            if (Objects.equals(change.getOldValue(), change.getNewValue())) {
              break;
            }
            if (change.getOldValue() != null) {
              change.setChangeType(PropertyChangeType.MODIFIED);
            }
            actualChanges.put(change.getPropertyName(), change);
            break;
          case MODIFIED:
            if (!Objects.equals(change.getOldValue(), change.getNewValue())) {
              actualChanges.put(change.getPropertyName(), change);
            }
            break;
          case DELETED:
            if (Objects.equals(change.getOldValue(), change.getNewValue())) {
              break;
            }
            if (change.getNewValue() != null) {
              change.setChangeType(PropertyChangeType.MODIFIED);
            }
            actualChanges.put(change.getPropertyName(), change);
            break;
          default:
            //do nothing
            break;
        }
      }
      return actualChanges.build();
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjI2MDI3MA_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/202001171721533.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjI2MDI3MA==,size_16,color_FFFFFF,t_70