Nacos源码系列—订阅机制的前因后果 | 下

点赞再看，养成习惯，微信搜索【牧小农】关注我获取更多资讯，风里雨里，小农等你，很高兴能够成为你的朋友。
项目源码地址：公众号回复 nacos，即可免费获取源码

事件发布

在上一节中我们讲解了在NotifyCenter中维护了事件名称和事件发布者的关系，而默认的事件发布者为DefaultPublisher，今天我们就来讲一下DefaultPublisher的事件发布的具体逻辑

首先我们来看一下DefaultPublisher的源码：

public class DefaultPublisher extends Thread implements EventPublisher {    @Override    public void init(Class<? extends Event> type, int bufferSize) {        //守护线程        setDaemon(true);        //设置线程名        setName("nacos.publisher-" + type.getName());        this.eventType = type;        this.queueMaxSize = bufferSize;        //阻塞队列初始化        this.queue = new ArrayBlockingQueue<>(bufferSize);        //启动线程        start();    }        @Override    public synchronized void start() {        if (!initialized) {            // start just called once            //启动run方法            super.start();            if (queueMaxSize == -1) {                queueMaxSize = ringBufferSize;            }            initialized = true;        }    }}

我们可以看到这个类继承自Thread，说明他是一个线程类，同时实现了EventPublisher说明他也是一个发布者，在init()中，是以守护线程的方式运作的，同时初始化了一个阻塞队列，最后调用start()启动线程。

在start()里面，其实就是启动run()：

@Overridepublic void run() {    openEventHandler();}   void openEventHandler() {    try {        // This variable is defined to resolve the problem which message overstock in the queue.        int waitTimes = 60;        // To ensure that messages are not lost, enable EventHandler when        // waiting for the first Subscriber to register        //死循环遍历，线程启动设置最大延迟60秒，用来解决消息积压问题        for (; ; ) {            if (shutdown || hasSubscriber() || waitTimes <= 0) {                break;            }            ThreadUtils.sleep(1000L);            waitTimes--;        }        //死循环从队列中取出event对象，同时通知订阅者(subscriber)执行event对象        for (; ; ) {            if (shutdown) {                break;            }            final Event event = queue.take();            receiveEvent(event);            UPDATER.compareAndSet(this, lastEventSequence, Math.max(lastEventSequence, event.sequence()));        }    } catch (Throwable ex) {        LOGGER.error("Event listener exception : ", ex);    }}

在上述代码中我们可以看到for (; ; )这个循环出现了两次，这个就是循环遍历(死循环)，第一个死循环我们可以理解成延时效果，里面最大延时60秒，每隔一秒运行一次，判断（当前线程是否关闭、是否有订阅者、是否超过60秒）只要满足其中任意一个条件，跳出循环
第二个死循环，是我们业务逻辑处理，用来消费，从队列中取出event事件，然后通过receiveEvent()执行。

那么我们可以从队列中取出事件，那么这个事件又在哪一步注入进去的呢，我们还是在当前类里面，找到一个叫publish()的方法

@Overridepublic boolean publish(Event event) {    checkIsStart();    //向队列中插入元素    boolean success = this.queue.offer(event);    //判断是否插入成功    if (!success) {        LOGGER.warn("Unable to plug in due to interruption, synchronize sending time, event : {}", event);        //失败直接执行        receiveEvent(event);        return true;    }    return true;}

这个方法其实就是发布事件调用了publish往阻塞队列中存入事件，如果失败那么立即执行receiveEvent()，不在继续走队列方法

void receiveEvent(Event event) {    final long currentEventSequence = event.sequence();    if (!hasSubscriber()) {        LOGGER.warn("[NotifyCenter] the {} is lost, because there is no subscriber.", event);        return;    }    // Notification single event listener    //循环遍历subscribers对象    for (Subscriber subscriber : subscribers) {        // Whether to ignore expiration events        if (subscriber.ignoreExpireEvent() && lastEventSequence > currentEventSequence) {            LOGGER.debug("[NotifyCenter] the {} is unacceptable to this subscriber, because had expire",                    event.getClass());            continue;        }        // Because unifying smartSubscriber and subscriber, so here need to think of compatibility.        // Remove original judge part of codes.        //通知订阅者执行event        notifySubscriber(subscriber, event);    }}

而在receiveEvent()方法中，这里其实就是遍历的subscribers集合（订阅者），然后通过notifySubscriber() 通知订阅者方法，而这个subscribers集合就是在我们之前讲到的NacosNamingService.init()方法中设置的。

public class NacosNamingService implements NamingService { private void init(Properties properties) throws NacosException {   //将Subscribe注册到Publisher   NotifyCenter.registerSubscriber(changeNotifier); }}

而 NotifyCenter.registerSubscriber(changeNotifier);会调用NotifyCenter.addSubscriber()方法，进行最终的操作。

private static void addSubscriber(final Subscriber consumer, Class<? extends Event> subscribeType,      EventPublisherFactory factory) {  final String topic = ClassUtils.getCanonicalName(subscribeType);  synchronized (NotifyCenter.class) {      // MapUtils.computeIfAbsent is a unsafe method.      MapUtil.computeIfAbsent(INSTANCE.publisherMap, topic, factory, subscribeType, ringBufferSize);  }  //获取对应的publisher  EventPublisher publisher = INSTANCE.publisherMap.get(topic);  if (publisher instanceof ShardedEventPublisher) {      ((ShardedEventPublisher) publisher).addSubscriber(consumer, subscribeType);  } else {      //添加到subscribers集合      publisher.addSubscriber(consumer);  }}

addSubscriber()方法的逻辑就是讲订阅事件、发布中、订阅者三个关系进行绑定，而发布者和事件通过Map进行维护，发布者与订阅者通过关联关系进行维护。

我们回到刚刚DefaulePublisher.notifySubscriber()方法，这里是最后执行订阅者事件的方法

@Overridepublic void notifySubscriber(final Subscriber subscriber, final Event event) {  LOGGER.debug("[NotifyCenter] the {} will received by {}", event, subscriber);  //执行订阅者事件  final Runnable job = () -> subscriber.onEvent(event);  //执行者  final Executor executor = subscriber.executor();  if (executor != null) {      executor.execute(job);  } else {      try {          job.run();      } catch (Throwable e) {          LOGGER.error("Event callback exception: ", e);      }  }}

到这里，订阅机制就讲完了，可能会有点绕，最好是我们能够去跟着代码走一遍，这样会比较理解和记忆，在这里我们重点需要理解NotifyCenter对事件发布者、订阅者以及之间关系的维护，关系维护的入口就在NacosNamingService.init()中，我们来看一下他的核心逻辑

首先ServiceInfoHolder中通过NotifyCenter发布InstancesChangeEvent事件.

NotifyCenter获取对应的CanonicalName，并将这个参数作为key，从NotifyCenter.publisherMap中获取对应的事件发布者，然后将InstancesChangeEvent事件进行发布.

InstancesChangeEvent事件发布主要是通过EventPublisher的实现类，DefaultPublisher进行InstancesChangeEvent事件发布，而DefaultPublisher本身作为守护线程的方式进行运作，在执行业务逻辑时判断是否线程启动，如果启动，将事件添加到队列中，如果成功，则发布过程完成，如果添加失败，立即执行DefaultPublisher.receiveEvent，接收事件通知订阅者，创建一个Runnable对象，执行订阅者的Event事件。

在添加到队列成功的时候，DefaultPublisher会创建一个阻塞队列(BlockingQueue)，标记线程启动，当他执行 super.start()，会调用它的run方法，在这个run方法里面核心的业务逻辑就是openEventHandler()，里面会有两个死循环，第一个是在线程启动的60秒内执行条件，第二个是从阻塞队列中获取Event事件，调用DefaultPublisher.receiveEvent()通知订阅者，流程结束

本地缓存

我们在之前的系列中，客户端会缓存一些信息在本地中，来获取ServiceInfo的信息，但是在执行本地缓存的时候，难免会有一些故障，有故障就需要进行处理，在这里主要涉及到两个类ServiceInfoHolder和FailoverReactor

Nacos缓存主要是分为两个方面，一个从注册中心获取实例信息缓存到内存中，通过ConcurrentMap进行存储，一个是通过磁盘文件的形式定时缓存。

同时故障处理也分为两个部分，一个是故障处理的开关通过文件进行标记，一个是当起来故障处理后，可以从故障备份的文件中获取服务实例信息。

介绍完上面几点，我们先来详细讲解第一个核心类ServiceInfoHolder

ServiceInfoHolder

ServiceInfoHolder类，主要是用来处理服务信息的，每次客户端从服务端拉取服务信息时，都用经过这个类，而processServiceInfo用来处理本地信息（缓存、发布、更新、本地目录初始化）等

ServiceInfo: 注册服务的信息，主要包含（服务名、分组名、集群信息、实例列表、最后一次更新时间），客户端获取的信息，都是通过ServiceInfo作为承载体，ServiceInfoHolder.ServiceInfo，通过ConcurrentMap进行存储，如下所示：

public class ServiceInfoHolder implements Closeable {  private final ConcurrentMap<String, ServiceInfo> serviceInfoMap;     public ServiceInfoHolder(String namespace, Properties properties) {    initCacheDir(namespace, properties);    //启动是判断是否从缓存信息中获取，默认为false    if (isLoadCacheAtStart(properties)) {        //从缓存目录中读取信息        this.serviceInfoMap = new ConcurrentHashMap<>(DiskCache.read(this.cacheDir));    } else {        //创建空集合对象        this.serviceInfoMap = new ConcurrentHashMap<>(16);    }    this.failoverReactor = new FailoverReactor(this, cacheDir);    this.pushEmptyProtection = isPushEmptyProtect(properties);}    public ServiceInfo processServiceInfo(ServiceInfo serviceInfo) {        //判断服务key是否为空        String serviceKey = serviceInfo.getKey();        if (serviceKey == null) {            return null;        }        ServiceInfo oldService = serviceInfoMap.get(serviceInfo.getKey());        if (isEmptyOrErrorPush(serviceInfo)) {            //empty or error push, just ignore            return oldService;        }        //将缓存信息放置到map中        serviceInfoMap.put(serviceInfo.getKey(), serviceInfo);        //判断实例信息是否发生改变        boolean changed = isChangedServiceInfo(oldService, serviceInfo);        if (StringUtils.isBlank(serviceInfo.getJsonFromServer())) {            serviceInfo.setJsonFromServer(JacksonUtils.toJson(serviceInfo));        }        //监控服务缓存map的大小        MetricsMonitor.getServiceInfoMapSizeMonitor().set(serviceInfoMap.size());        if (changed) {            NAMING_LOGGER.info("current ips:({}) service: {} -> {}", serviceInfo.ipCount(), serviceInfo.getKey(),                    JacksonUtils.toJson(serviceInfo.getHosts()));            //添加实例变更事件，被订阅者执行            NotifyCenter.publishEvent(new InstancesChangeEvent(serviceInfo.getName(), serviceInfo.getGroupName(),                    serviceInfo.getClusters(), serviceInfo.getHosts()));            //写入本地文件            DiskCache.write(serviceInfo, cacheDir);        }        return serviceInfo;    }}

这里就是Nacos获取注册信息的缓存，之前我们有讲过，当服务信息变更的时候会第一时间更新ServiceInfoMap中的信息，通过isChangedServiceInfo进行判断，当发生变动时，serviceInfoMap.put最新数据，当我们需要使用的时候，通过key进行get操作，ServiceInfoMap默认创建空的对象，但如果配置启动从缓存文件中获取，则会从缓存中获取信息。而且当我们服务实例发生变更的时候，会通过DiskCache.write()向对应的目录文件中写入ServiceInfo信息

本地缓存地址

本地缓存的地址通过cacheDir进行执行本地缓存和故障处理的根目录，在ServiceInfoHolder构造方法中，会默认生成缓存目录，默认路径为${user}/nacos/naming/public，我们也可以需要通过System.setProperty("JM.SNAPSHOT.PATH")指定。

public class ServiceInfoHolder implements Closeable {    private String cacheDir;        public ServiceInfoHolder(String namespace, Properties properties) {    //初始化生成缓存目录    initCacheDir(namespace, properties);    ......    }        private void initCacheDir(String namespace, Properties properties) {        String jmSnapshotPath = System.getProperty(JM_SNAPSHOT_PATH_PROPERTY);        String namingCacheRegistryDir = "";        if (properties.getProperty(PropertyKeyConst.NAMING_CACHE_REGISTRY_DIR) != null) {            namingCacheRegistryDir = File.separator + properties.getProperty(PropertyKeyConst.NAMING_CACHE_REGISTRY_DIR);        }        if (!StringUtils.isBlank(jmSnapshotPath)) {            cacheDir = jmSnapshotPath + File.separator + FILE_PATH_NACOS + namingCacheRegistryDir                    + File.separator + FILE_PATH_NAMING + File.separator + namespace;        } else {            cacheDir = System.getProperty(USER_HOME_PROPERTY) + File.separator + FILE_PATH_NACOS + namingCacheRegistryDir                    + File.separator + FILE_PATH_NAMING + File.separator + namespace;        }    }    }

故障处理

在ServiceInfoHolder构造方法中，还会初始化一个FailoverReactor的类，这个类主要是用来故障处理。

public class ServiceInfoHolder implements Closeable {  private final FailoverReactor failoverReactor;    public ServiceInfoHolder(String namespace, Properties properties) {    ....    //为两者相互持有对方的引用    this.failoverReactor = new FailoverReactor(this, cacheDir);    .....  }     public FailoverReactor(ServiceInfoHolder serviceInfoHolder, String cacheDir) {        //获取serviceInfoHolder引用对象        this.serviceInfoHolder = serviceInfoHolder;        //故障目录${user}/nacos/naming/public/failover        this.failoverDir = cacheDir + FAILOVER_DIR;        //初始化executorService        this.executorService = new ScheduledThreadPoolExecutor(1, new ThreadFactory() {            @Override            public Thread newThread(Runnable r) {                Thread thread = new Thread(r);                //开启守护线程                thread.setDaemon(true);                thread.setName("com.alibaba.nacos.naming.failover");                return thread;            }        });        //其他信息初始化        this.init();    }         public void init() {        //执行初始化操作，间隔5秒，执行SwitchRefresher()任务        executorService.scheduleWithFixedDelay(new SwitchRefresher(), 0L, 5000L, TimeUnit.MILLISECONDS);        //初始化操作，延迟30分钟执行，间隔24小时，执行DiskFileWriter()任务        executorService.scheduleWithFixedDelay(new DiskFileWriter(), 30, DAY_PERIOD_MINUTES, TimeUnit.MINUTES);                //初始化操作，间隔10秒，核心方法为DiskFileWriter        executorService.schedule(new Runnable() {            @Override            public void run() {                try {                    File cacheDir = new File(failoverDir);                                        if (!cacheDir.exists() && !cacheDir.mkdirs()) {                        throw new IllegalStateException("failed to create cache dir: " + failoverDir);                    }                                        File[] files = cacheDir.listFiles();                    //如果故障目录为空，启动立即执行，备份文件                    if (files == null || files.length <= 0) {                        new DiskFileWriter().run();                    }                } catch (Throwable e) {                    NAMING_LOGGER.error("[NA] failed to backup file on startup.", e);                }                            }        }, 10000L, TimeUnit.MILLISECONDS);    }}

在init()代码中，开启了三个定时任务，三个任务都是FailoverReactor内部类，

• 执行初始化操作，间隔5秒，执行SwitchRefresher()任务
• 初始化操作，延迟30分钟执行，间隔24小时，执行DiskFileWriter()任务
• 初始化操作，间隔10秒，核心方法为DiskFileWriter

我们这里先来看一下核心方法DiskFileWriter,这里主要是获取服务信息，判断是否能够写入磁盘，条件满足，写入拼接的故障目录中，因为第一个和第二个初始化操作，都会用到DiskFileWriter，当我们第三个定时判断如果文件不存在，则会将文件写入本地磁盘中

class DiskFileWriter extends TimerTask {  @Override  public void run() {      Map<String, ServiceInfo> map = serviceInfoHolder.getServiceInfoMap();      for (Map.Entry<String, ServiceInfo> entry : map.entrySet()) {          ServiceInfo serviceInfo = entry.getValue();          //主要是判断服务信息是否完整          if (StringUtils.equals(serviceInfo.getKey(), UtilAndComs.ALL_IPS) || StringUtils                  .equals(serviceInfo.getName(), UtilAndComs.ENV_LIST_KEY) || StringUtils                  .equals(serviceInfo.getName(), UtilAndComs.ENV_CONFIGS) || StringUtils                  .equals(serviceInfo.getName(), UtilAndComs.VIP_CLIENT_FILE) || StringUtils                  .equals(serviceInfo.getName(), UtilAndComs.ALL_HOSTS)) {              continue;          }          //将文件写入磁盘中          DiskCache.write(serviceInfo, failoverDir);      }  }}

接下来我们再看一下，第一个定时任务SwitchRefresher的业务逻辑，

class SwitchRefresher implements Runnable {long lastModifiedMillis = 0L;@Overridepublic void run() {  try {      File switchFile = new File(failoverDir + UtilAndComs.FAILOVER_SWITCH);      //如果文件不存在返回      if (!switchFile.exists()) {          switchParams.put(FAILOVER_MODE_PARAM, Boolean.FALSE.toString());          NAMING_LOGGER.debug("failover switch is not found, {}", switchFile.getName());          return;      }      long modified = switchFile.lastModified();      //判断文件修改时间      if (lastModifiedMillis < modified) {          lastModifiedMillis = modified;          //获取故障处理文件内容          String failover = ConcurrentDiskUtil.getFileContent(failoverDir + UtilAndComs.FAILOVER_SWITCH,                  Charset.defaultCharset().toString());          if (!StringUtils.isEmpty(failover)) {              String[] lines = failover.split(DiskCache.getLineSeparator());              for (String line : lines) {                  String line1 = line.trim();                  //"1" 开启故障处理                  if (IS_FAILOVER_MODE.equals(line1)) {                      switchParams.put(FAILOVER_MODE_PARAM, Boolean.TRUE.toString());                      NAMING_LOGGER.info("failover-mode is on");                      new FailoverFileReader().run();                      //"0" 关闭故障处理                  } else if (NO_FAILOVER_MODE.equals(line1)) {                      switchParams.put(FAILOVER_MODE_PARAM, Boolean.FALSE.toString());                      NAMING_LOGGER.info("failover-mode is off");                  }              }          } else {              switchParams.put(FAILOVER_MODE_PARAM, Boolean.FALSE.toString());          }      }  } catch (Throwable e) {      NAMING_LOGGER.error("[NA] failed to read failover switch.", e);  }}}

这里面主要是判断故障处理文件是否存在，不存在直接返回，再去比较文件的修改时间，如果已经修改，则获取文件中的内容，继续进行判断，当我们开启故障处理时，执行线程FailoverFileReader().run()

class FailoverFileReader implements Runnable {@Overridepublic void run() {  Map<String, ServiceInfo> domMap = new HashMap<String, ServiceInfo>(16);  BufferedReader reader = null;  try {      //读取failoverDir目录下的文件      File cacheDir = new File(failoverDir);      //不存在返回错误      if (!cacheDir.exists() && !cacheDir.mkdirs()) {          throw new IllegalStateException("failed to create cache dir: " + failoverDir);      }      //获取文件      File[] files = cacheDir.listFiles();      //文件不存在返回      if (files == null) {          return;      }      //遍历处理      for (File file : files) {          //文件不存在跳过          if (!file.isFile()) {              continue;          }          //如果是故障处理标志文件，跳过这一步          if (file.getName().equals(UtilAndComs.FAILOVER_SWITCH)) {              continue;          }          ServiceInfo dom = new ServiceInfo(file.getName());          //读取备份中的内容，转换为ServiceInfo对象          try {              String dataString = ConcurrentDiskUtil                      .getFileContent(file, Charset.defaultCharset().toString());              reader = new BufferedReader(new StringReader(dataString));              String json;              if ((json = reader.readLine()) != null) {                  try {                      dom = JacksonUtils.toObj(json, ServiceInfo.class);                  } catch (Exception e) {                      NAMING_LOGGER.error("[NA] error while parsing cached dom : {}", json, e);                  }              }          } catch (Exception e) {              NAMING_LOGGER.error("[NA] failed to read cache for dom: {}", file.getName(), e);          } finally {              try {                  if (reader != null) {                      reader.close();                  }              } catch (Exception e) {                  //ignore              }          }          if (!CollectionUtils.isEmpty(dom.getHosts())) {              //将ServiceInfo对象放入domMap中              domMap.put(dom.getKey(), dom);          }      }  } catch (Exception e) {      NAMING_LOGGER.error("[NA] failed to read cache file", e);  }  //如果不为空，赋值serviceMap  if (domMap.size() > 0) {      serviceMap = domMap;  }}}

FailoverFileReader主要是操作读取failover目录存储的备份服务信息文件内容，然后装换成ServiceInfo信息，并将所有的ServiceInfo储存在FailoverReactor的ServiceMap属性中。

总结

到这里我们Nacos订阅机制核心流程就讲完了，整体订阅机制的流程还是比较复杂的，因为还涉及到之前将的逻辑，会有点绕，并且用到了保证线程Map、守护线程、阻塞队列、线程的使用等等，我们需要重点掌握的主要是事件发布者、订阅者之间的关系，这里还是推荐大家有机会的话可以自己跟着源码走一遍，会有更深的体验。

如果觉得文中有帮助的，记得点赞支持，你的支持是我创作的最大动力！

我是牧小农，怕什么真理无穷，进一步有进一步的欢喜，大家加油！