学习不必那么功利，二师兄带你从更高维度轻松浏览源码～
上篇文章，咱们剖析了Nacos客户端订阅的外围流程：Nacos客户端通过一个定时工作，每6秒从注册核心获取实例列表，当发现实例发生变化时，公布变更事件，订阅者进行业务解决，而后更新内存中和本地的缓存中的实例。
这篇文章为服务订阅的第二篇，咱们重点来剖析，定时工作获取到最新实例列表之后，整个事件机制是如何解决的。
回顾整个流程
先回顾一下客户端服务订阅的根本流程：
在第一步调用subscribe办法时，会订阅一个EventListener事件。而在定时工作UpdateTask定时获取实例列表之后，会调用ServiceInfoHolder#processServiceInfo办法对ServiceInfo进行本地解决，这其中就包含和事件处理。
监听事件的注册
在subscribe办法中，通过如下形式进行了监听事件的注册：
@Overridepublic void subscribe(String serviceName, String groupName, List<String> clusters, EventListener listener) throws NacosException { if (null == listener) { return; } String clusterString = StringUtils.join(clusters, ","); changeNotifier.registerListener(groupName, serviceName, clusterString, listener); clientProxy.subscribe(serviceName, groupName, clusterString);}这里的changeNotifier.registerListener便是进行具体的事件注册逻辑。追进去看一下实现源码：
// InstancesChangeNotifierpublic void registerListener(String groupName, String serviceName, String clusters, EventListener listener) { String key = ServiceInfo.getKey(NamingUtils.getGroupedName(serviceName, groupName), clusters); ConcurrentHashSet<EventListener> eventListeners = listenerMap.get(key); if (eventListeners == null) { synchronized (lock) { eventListeners = listenerMap.get(key); if (eventListeners == null) { eventListeners = new ConcurrentHashSet<EventListener>(); // 将EventListener缓存到listenerMap listenerMap.put(key, eventListeners); } } } eventListeners.add(listener);}能够看出，事件的注册便是将EventListener存储在InstancesChangeNotifier的listenerMap属性当中了。
这里的数据结构为Map，key为服务实例信息的拼接，value为监听事件的汇合。
事件注册流程就这么简略。这里有一个双重查看锁的实际案例，不晓得你留意到没？能够学习一下。
ServiceInfo的解决
下面实现了事件的注册，当初就追溯一下触发事件的起源。UpdateTask中获取到最新实例会进行本地化解决，局部代码如下：
// 获取缓存的service信息ServiceInfo serviceObj = serviceInfoHolder.getServiceInfoMap().get(serviceKey);if (serviceObj == null) { // 依据serviceName从注册核心服务端获取Service信息 serviceObj = namingClientProxy.queryInstancesOfService(serviceName, groupName, clusters, 0, false); serviceInfoHolder.processServiceInfo(serviceObj); lastRefTime = serviceObj.getLastRefTime(); return;}这部分逻辑在上篇文章中曾经剖析过了，这里重点看serviceInfoHolder#processServiceInfo中的业务逻辑解决。先看流程图，而后看代码。
上述逻辑简略说就是：判断一下新的ServiceInfo数据是否正确，是否产生了变动。如果数据格式正确，且产生的变动，那就公布一个InstancesChangeEvent事件，同时将ServiceInfo写入本地缓存。
上面看一下代码实现：
public ServiceInfo processServiceInfo(ServiceInfo serviceInfo) { 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; } // 缓存服务信息 serviceInfoMap.put(serviceInfo.getKey(), serviceInfo); // 判断注册的实例信息是否已变更 boolean changed = isChangedServiceInfo(oldService, serviceInfo); if (StringUtils.isBlank(serviceInfo.getJsonFromServer())) { serviceInfo.setJsonFromServer(JacksonUtils.toJson(serviceInfo)); } // 通过prometheus-simpleclient监控服务缓存Map的大小 MetricsMonitor.getServiceInfoMapSizeMonitor().set(serviceInfoMap.size()); // 服务实例已变更 if (changed) { NAMING_LOGGER.info("current ips:(" + serviceInfo.ipCount() + ") service: " + serviceInfo.getKey() + " -> " + JacksonUtils.toJson(serviceInfo.getHosts())); // 增加实例变更事件，会被推动到订阅者执行 NotifyCenter.publishEvent(new InstancesChangeEvent(serviceInfo.getName(), serviceInfo.getGroupName(), serviceInfo.getClusters(), serviceInfo.getHosts())); // 记录Service本地文件 DiskCache.write(serviceInfo, cacheDir); } return serviceInfo;}能够对照流程图和代码中的正文局部进行了解这个过程。
咱们要讲的重点是服务信息变更之后，公布的InstancesChangeEvent，也就是流程图中标红的局部。
事件追踪
下面的事件是通过NotifyCenter进行公布的，NotifyCenter中的外围流程如下：
NotifyCenter中进行事件公布，公布的外围逻辑是：
- 依据InstancesChangeEvent事件类型，取得对应的CanonicalName；
- 将CanonicalName作为Key，从NotifyCenter#publisherMap中获取对应的事件发布者（EventPublisher）；
- EventPublisher将InstancesChangeEvent事件进行公布。
NotifyCenter中的外围代码实现如下：
private static boolean publishEvent(final Class<? extends Event> eventType, final Event event) { if (ClassUtils.isAssignableFrom(SlowEvent.class, eventType)) { return INSTANCE.sharePublisher.publish(event); } // 依据InstancesChangeEvent事件类型，取得对应的CanonicalName； final String topic = ClassUtils.getCanonicalName(eventType); // 将CanonicalName作为Key，从NotifyCenter#publisherMap中获取对应的事件发布者（EventPublisher）； EventPublisher publisher = INSTANCE.publisherMap.get(topic); if (publisher != null) { // EventPublisher将InstancesChangeEvent事件进行公布。 return publisher.publish(event); } LOGGER.warn("There are no [{}] publishers for this event, please register", topic); return false;}上述代码中的INSTANCE为NotifyCenter的单例模式实现。那么，这里的publisherMap中key（CanonicalName）和value（EventPublisher）之间的关系是什么时候建设的呢？
这个是在NacosNamingService实例化时调用init办法中进行绑定的：
// Publisher的注册过程在于建设InstancesChangeEvent.class与EventPublisher的关系。NotifyCenter.registerToPublisher(InstancesChangeEvent.class, 16384);registerToPublisher办法默认采纳了DEFAULT_PUBLISHER_FACTORY来进行构建。
public static EventPublisher registerToPublisher(final Class<? extends Event> eventType, final int queueMaxSize) { return registerToPublisher(eventType, DEFAULT_PUBLISHER_FACTORY, queueMaxSize);}如果查看NotifyCenter中动态代码块，会发现DEFAULT_PUBLISHER_FACTORY默认构建的EventPublisher为DefaultPublisher。
至此，咱们得悉，在NotifyCenter中它保护了事件名称和事件发布者的关系，而默认的事件发布者为DefaultPublisher。
DefaultPublisher的事件公布
查看DefaultPublisher的源码，会发现它继承自Thread，也就是说它是一个线程类。同时，它又实现了EventPublisher，也就是咱们后面提到的发布者。
public class DefaultPublisher extends Thread implements EventPublisher {}在DefaultPublisher的init办法实现如下：
@Overridepublic 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();}也就是说，当DefaultPublisher被初始化时，是以守护线程的形式运作的，其中还初始化了一个阻塞队列，队列的默认大小为16384。
最初调用了start办法：
@Overridepublic synchronized void start() { if (!initialized) { // start just called once super.start(); if (queueMaxSize == -1) { queueMaxSize = ringBufferSize; } initialized = true; }}start办法中调用了super.start，此时等于启动了线程，会执行对应的run办法。
run办法中只调用了如下办法：
void openEventHandler() { try { // This variable is defined to resolve the problem which message overstock in the queue. int waitTimes = 60; // for死循环不断的从队列中取出Event，并告诉订阅者Subscriber执行Event // To ensure that messages are not lost, enable EventHandler when // waiting for the first Subscriber to register for (; ; ) { if (shutdown || hasSubscriber() || waitTimes <= 0) { break; } ThreadUtils.sleep(1000L); waitTimes--; } for (; ; ) { if (shutdown) { break; } // // 从队列取出Event 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); }}这里写了两个死循环，第一个死循环能够了解为延时成果，也就是说线程启动时最大延时60秒，在这60秒中每隔1秒判断一下以后线程是否敞开，是否有订阅者，是否超过60秒。如果满足一个条件，就能够提前跳出死循环。
而第二个死循环才是真正的业务逻辑解决，会从阻塞队列中取出一个事件，而后通过receiveEvent办法进行执行。
那么，队列中的事件哪儿来的呢？此时，你可能曾经想到方才DefaultPublisher的公布事件办法被调用了。来看看它的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;}能够看到，DefaultPublisher的publish办法确实就是往阻塞队列中存入事件。这里有个分支逻辑，如果存入失败，会间接调用receiveEvent，和从队列中取出事件执行的办法一样。能够了解为，如果向队列中存入失败，则立刻执行，不走队列了。
最初，再来看看receiveEvent办法的实现：
void receiveEvent(Event event) { final long currentEventSequence = event.sequence(); if (!hasSubscriber()) { LOGGER.warn("[NotifyCenter] the {} is lost, because there is no subscriber."); return; } // 告诉订阅者执行Event // Notification single event listener 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. notifySubscriber(subscriber, event); }}这里最次要的逻辑就是遍历DefaultPublisher的subscribers（订阅者汇合），而后执行告诉订阅者的办法。
那么有敌人要问了这subscribers中的订阅者哪里来的呢？这个还要回到NacosNamingService的init办法中：
// 将Subscribe注册到PublisherNotifyCenter.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); }}其中外围逻辑就是将订阅事件、发布者、订阅者三者进行绑定。而发布者与事件通过Map进行保护、发布者与订阅者通过关联关系进行保护。
发布者找到了，事件也有了，最初看一下notifySubscriber办法：
@Overridepublic void notifySubscriber(final Subscriber subscriber, final Event event) { LOGGER.debug("[NotifyCenter] the {} will received by {}", event, subscriber); // 执行订阅者Event 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); } }}逻辑比较简单，如果订阅者定义了Executor，那么应用它定义的Executor进行事件的执行，如果没有，那就创立一个线程进行执行。
至此，整个服务订阅的事件机制实现。
小结
整体来看，整个服务订阅的事件机制还是比较复杂的，因为用到了事件的模式，逻辑就比拟绕，而且这期间还掺杂了守护线程，死循环，阻塞队列等。须要重点了解NotifyCenter对事件发布者、事件订阅者和事件之间关系的保护，而这一关系的保护的入口就位于NacosNamingService的init办法当中。
上面再梳理一下几个外围流程：
ServiceInfoHolder中通过NotifyCenter公布了InstancesChangeEvent事件；
NotifyCenter中进行事件公布，公布的外围逻辑是：
- 依据InstancesChangeEvent事件类型，取得对应的CanonicalName；
- 将CanonicalName作为Key，从NotifyCenter#publisherMap中获取对应的事件发布者（EventPublisher）；
- EventPublisher将InstancesChangeEvent事件进行公布。
InstancesChangeEvent事件公布：
- 通过EventPublisher的实现类DefaultPublisher进行InstancesChangeEvent事件公布；
- DefaultPublisher自身以守护线程的形式运作，在执行业务逻辑前，先判断该线程是否启动；
- 如果启动，则将事件增加到BlockingQueue中，队列默认大小为16384；
- 增加到BlockingQueue胜利，则整个公布过程实现；
- 如果增加失败，则间接调用DefaultPublisher#receiveEvent办法，接管事件并告诉订阅者；
- 告诉订阅者时创立一个Runnable对象，执行订阅者的Event。
- Event事件便是执行订阅时传入的事件；
如果增加到BlockingQueue胜利，则走另外一个业务逻辑：
- DefaultPublisher初始化时会创立一个阻塞（BlockingQueue）队列，并标记线程启动；
- DefaultPublisher自身是一个Thread，当执行super.start办法时，会调用它的run办法；
- run办法的外围业务逻辑是通过openEventHandler办法解决的；
- openEventHandler办法通过两个for循环，从阻塞队列中获取工夫信息；
- 第一个for循环用于让线程启动时在60s内查看执行条件；
- 第二个for循环为死循环，从阻塞队列中获取Event，并调用DefaultPublisher#receiveEvent办法，接管事件并告诉订阅者；
- Event事件便是执行订阅时传入的事件；
对于Nacos Client服务定义的事件机制就将这么多，下篇咱们来讲讲故障转移和缓存的实现。
博主简介：《SpringBoot技术底细》技术图书作者，热爱钻研技术，写技术干货文章。
公众号：「程序新视界」，博主的公众号，欢送关注~
技术交换：请分割博主微信号：zhuan2quan