10 Nacos配置中心之事件订阅

10 Nacos配置中心之事件订阅

我们都知道，Nacos配置中心支持动态的刷新，有没有想过它是怎么实现的呢?

当配置中心的配置发送
NacosContextRefresher实现ApplicationListener接口进行事件监听，在上下文准备完毕时候触发这个事件

NacosContextRefresher的onApplicationEvent方法：

@Override
public void onApplicationEvent(ApplicationReadyEvent event) {
   // many Spring context
   if (this.ready.compareAndSet(false, true)) {
      this.registerNacosListenersForApplications();
   }
}

使用CAS加锁，把 ready变量改为true

调用registerNacosListenersForApplications()进行Nacos事件监听的注册。

registerNacosListenersForApplications方法：

private void registerNacosListenersForApplications() {
   if (refreshProperties.isEnabled()) {
      for (NacosPropertySource nacosPropertySource : NacosPropertySourceRepository
            .getAll()) {

         if (!nacosPropertySource.isRefreshable()) {
            continue;
         }

         String dataId = nacosPropertySource.getDataId();
         registerNacosListener(nacosPropertySource.getGroup(), dataId);
      }
   }
}

1. 获取dataId
2. 如果属性源不支持刷新就跳过，支持就调用registerNacosListener()方法，我们看一下这个方法

registerNacosListener()方法

private void registerNacosListener(final String group, final String dataId) {

   Listener listener = listenerMap.computeIfAbsent(dataId, i -> new Listener() {
      @Override
      public void receiveConfigInfo(String configInfo) {
         refreshCountIncrement();
         String md5 = "";
         if (!StringUtils.isEmpty(configInfo)) {
            try {
               MessageDigest md = MessageDigest.getInstance("MD5");
               md5 = new BigInteger(1, md.digest(configInfo.getBytes("UTF-8")))
                     .toString(16);
            }
            catch (NoSuchAlgorithmException | UnsupportedEncodingException e) {
               log.warn("[Nacos] unable to get md5 for dataId: " + dataId, e);
            }
         }
         refreshHistory.add(dataId, md5);
         applicationContext.publishEvent(
               new RefreshEvent(this, null, "Refresh Nacos config"));
         if (log.isDebugEnabled()) {
            log.debug("Refresh Nacos config group " + group + ",dataId" + dataId);
         }
      }

      @Override
      public Executor getExecutor() {
         return null;
      }
   });

   try {
      configService.addListener(dataId, group, listener);
   }
   catch (NacosException e) {
      e.printStackTrace();
   }
}

1. 创建监听器添加到listenerMap集合中，监听器中主要刷新个数，添加dataId和md5到监听历史中，最后发布监听事件
2. 把监听器在添加到configService中，这篇文章我们有介绍这个类，它是Nacos客户端提供的用于访问实现配置中心基本操作的类

registerNacosListener中applicationContext.publishEvent方法发布RefreshEvent事件，监听这个事件到实现类在RefreshEventListener类中

事件监听

RefreshEventListener类：

public class RefreshEventListener implements SmartApplicationListener {
     private ContextRefresher refresh;

    public void onApplicationEvent(ApplicationEvent event) {
        if (event instanceof ApplicationReadyEvent) {
            this.handle((ApplicationReadyEvent)event);
        } else if (event instanceof RefreshEvent) {
            this.handle((RefreshEvent)event);
        }

    }
    public void handle(RefreshEvent event) {
        if (this.ready.get()) {
            log.debug("Event received " + event.getEventDesc());
            Set<String> keys = this.refresh.refresh();
            log.info("Refresh keys changed: " + keys);
        }

    }

handler方法中调用refresh.refresh()方法完成配置的更新和应用。

总结

现在我们知道Nacos配置中心的时间订阅是怎么回事了，首先NacosContextRefresher实现了ApplicationListener接口，在上下文准备完毕时候触发事件方法，方法中调用refresh.refresh()方法来完成配置的更新和应用,对于事件订阅，这里用到的设计模式属于是观察者模式，所以说设计模式在我们的源码中无处不在，阅读源码能让我们学习好的思想，让自己更好的解决工作中的问题，下篇文章中我们将介绍nacos的客户端的长轮询机制是怎么工作的，我们下篇文章见吧~