nacos订阅者（nacos主动下线服务）

学习不用那么功利，二师兄带你从更高维度轻松阅读源码～

上篇文章，我们分析了Nacos客户端订阅的核心流程：Nacos客户端通过一个定时任务，每6秒从注册中心获取实例列表，当发现实例发生变化时，发布变更事件，订阅者进行业务处理，然后更新内存中和本地的缓存中的实例。

这篇文章为服务订阅的第二篇，我们重点来分析，定时任务获取到最新实例列表之后，整个事件机制是如何处理的。

回顾整个流程

先回顾一下客户端服务订阅的基本流程：

在第一步调用subscribe方法时，会订阅一个EventListener事件。而在定时任务UpdateTask定时获取实例列表之后，会调用ServiceInfoHolder#processServiceInfo方法对ServiceInfo进行本地处理，这其中就包括和事件处理。

监听事件的注册

在subscribe方法中，通过如下方式进行了监听事件的注册：

1. @Override
2. publicvoidsubscribe(StringserviceName,StringgroupName,List<String>clusters,EventListenerlistener)
3. throwsNacosException{
4. if(null==listener){
5. return;
6. }
7. StringclusterString=StringUtils.join(clusters,",");
8. changeNotifier.registerListener(groupName,serviceName,clusterString,listener);
9. clientProxy.subscribe(serviceName,groupName,clusterString);
10. }

这里的changeNotifier.registerListener便是进行具体的事件注册逻辑。追进去看一下实现源码：

1. //InstancesChangeNotifier
2. publicvoidregisterListener(StringgroupName,StringserviceName,Stringclusters,EventListenerlistener){
3. Stringkey=ServiceInfo.getKey(NamingUtils.getGroupedName(serviceName,groupName),clusters);
4. ConcurrentHashSet<EventListener>eventListeners=listenerMap.get(key);
5. if(eventListeners==null){
6. synchronized(lock){
7. eventListeners=listenerMap.get(key);
8. if(eventListeners==null){
9. eventListeners=newConcurrentHashSet<EventListener>();
10. //将EventListener缓存到listenerMap
11. listenerMap.put(key,eventListeners);
12. }
13. }
14. }
15. eventListeners.add(listener);
16. }

可以看出，事件的注册便是将EventListener存储在InstancesChangeNotifier的listenerMap属性当中了。

这里的数据结构为Map，key为服务实例信息的拼接，value为监听事件的集合。

事件注册流程就这么简单。这里有一个双重检查锁的实践案例，不知道你留意到没?可以学习一下。

ServiceInfo的处理

上面完成了事件的注册，现在就追溯一下触发事件的来源。UpdateTask中获取到最新实例会进行本地化处理，部分代码如下：

1. //获取缓存的service信息
2. ServiceInfoserviceObj=serviceInfoHolder.getServiceInfoMap().get(serviceKey);
3. if(serviceObj==null){
4. //根据serviceName从注册中心服务端获取Service信息
5. serviceObj=namingClientProxy.queryInstancesOfService(serviceName,groupName,clusters,0,false);
6. serviceInfoHolder.processServiceInfo(serviceObj);
7. lastRefTime=serviceObj.getLastRefTime();
8. return;
9. }

这部分逻辑在上篇文章中已经分析过了，这里重点看serviceInfoHolder#processServiceInfo中的业务逻辑处理。先看流程图，然后看代码。

上述逻辑简单说就是：判断一下新的ServiceInfo数据是否正确，是否发生了变化。如果数据格式正确，且发生的变化，那就发布一个InstancesChangeEvent事件，同时将ServiceInfo写入本地缓存。

下面看一下代码实现：

1. publicServiceInfoprocessServiceInfo(ServiceInfoserviceInfo){
2. StringserviceKey=serviceInfo.getKey();
3. if(serviceKey==null){
4. returnnull;
5. }
6. ServiceInfooldService=serviceInfoMap.get(serviceInfo.getKey());
7. if(isEmptyOrErrorPush(serviceInfo)){
8. //emptyorerrorpush,justignore
9. returnoldService;
10. }
11. //缓存服务信息
12. serviceInfoMap.put(serviceInfo.getKey(),serviceInfo);
13. //判断注册的实例信息是否已变更
14. booleanchanged=isChangedServiceInfo(oldService,serviceInfo);
15. if(StringUtils.isBlank(serviceInfo.getJsonFromServer())){
16. serviceInfo.setJsonFromServer(JacksonUtils.toJson(serviceInfo));
17. }
18. //通过prometheus-simpleclient监控服务缓存Map的大小
19. MetricsMonitor.getServiceInfoMapSizeMonitor().set(serviceInfoMap.size());
20. //服务实例已变更
21. if(changed){
22. NAMING_LOGGER.info("currentips:("+serviceInfo.ipCount()+")service:"+serviceInfo.getKey()+"->"
23. +JacksonUtils.toJson(serviceInfo.getHosts()));
24. //添加实例变更事件，会被推动到订阅者执行
25. NotifyCenter.publishEvent(newInstancesChangeEvent(serviceInfo.getName(),serviceInfo.getGroupName(),
26. serviceInfo.getClusters(),serviceInfo.getHosts()));
27. //记录Service本地文件
28. DiskCache.write(serviceInfo,cacheDir);
29. }
30. returnserviceInfo;
31. }

可以对照流程图和代码中的注释部分进行理解这个过程。

我们要讲的重点是服务信息变更之后，发布的InstancesChangeEvent，也就是流程图中标红的部分。

事件追踪

上面的事件是通过NotifyCenter进行发布的，NotifyCenter中的核心流程如下：

NotifyCenter中进行事件发布，发布的核心逻辑是：

• 根据InstancesChangeEvent事件类型，获得对应的CanonicalName;
• 将CanonicalName作为Key，从NotifyCenter#publisherMap中获取对应的事件发布者(EventPublisher);
• EventPublisher将InstancesChangeEvent事件进行发布。

NotifyCenter中的核心代码实现如下：

1. privatestaticbooleanpublishEvent(finalClass<?extendsEvent>eventType,finalEventevent){
2. if(ClassUtils.isAssignableFrom(SlowEvent.class,eventType)){
3. returnINSTANCE.sharePublisher.publish(event);
4. }
6. //根据InstancesChangeEvent事件类型，获得对应的CanonicalName；
7. finalStringtopic=ClassUtils.getCanonicalName(eventType);
9. //将CanonicalName作为Key，从NotifyCenter#publisherMap中获取对应的事件发布者（EventPublisher）；
10. EventPublisherpublisher=INSTANCE.publisherMap.get(topic);
11. if(publisher!=null){
12. //EventPublisher将InstancesChangeEvent事件进行发布。
13. returnpublisher.publish(event);
14. }
15. LOGGER.warn("Thereareno[{}]publishersforthisevent,pleaseregister",topic);
16. returnfalse;
17. }

上述代码中的INSTANCE为NotifyCenter的单例模式实现。那么，这里的publisherMap中key(CanonicalName)和value(EventPublisher)之间的关系是什么时候建立的呢?

这个是在NacosNamingService实例化时调用init方法中进行绑定的：

1. //Publisher的注册过程在于建立InstancesChangeEvent.class与EventPublisher的关系。
2. NotifyCenter.registerToPublisher(InstancesChangeEvent.class,16384);

registerToPublisher方法默认采用了DEFAULT_PUBLISHER_FACTORY来进行构建。

1. publicstaticEventPublisherregisterToPublisher(finalClass<?extendsEvent>eventType,finalintqueueMaxSize){
2. returnregisterToPublisher(eventType,DEFAULT_PUBLISHER_FACTORY,queueMaxSize);
3. }

如果查看NotifyCenter中静态代码块，会发现DEFAULT_PUBLISHER_FACTORY默认构建的EventPublisher为DefaultPublisher。

至此，我们得知，在NotifyCenter中它维护了事件名称和事件发布者的关系，而默认的事件发布者为DefaultPublisher。

DefaultPublisher的事件发布

查看DefaultPublisher的源码，会发现它继承自Thread，也就是说它是一个线程类。同时，它又实现了EventPublisher，也就是我们前面提到的发布者。

1. publicclassDefaultPublisherextendsThreadimplementsEventPublisher{}

在DefaultPublisher的init方法实现如下：

1. @Override
2. publicvoidinit(Class<?extendsEvent>type,intbufferSize){
3. //守护线程
4. setDaemon(true);
5. //设置线程名字
6. setName("nacos.publisher-"+type.getName());
7. this.eventType=type;
8. this.queueMaxSize=bufferSize;
9. //阻塞队列初始化
10. this.queue=newArrayBlockingQueue<>(bufferSize);
11. start();
12. }

也就是说，当DefaultPublisher被初始化时，是以守护线程的方式运作的，其中还初始化了一个阻塞队列，队列的默认大小为16384。

最后调用了start方法：

1. @Override
2. publicsynchronizedvoidstart(){
3. if(!initialized){
4. //startjustcalledonce
5. super.start();
6. if(queueMaxSize==-1){
7. queueMaxSize=ringBufferSize;
8. }
9. initialized=true;
10. }
11. }

start方法中调用了super.start，此时等于启动了线程，会执行对应的run方法。

run方法中只调用了如下方法：

1. voidopenEventHandler(){
2. try{
4. //Thisvariableisdefinedtoresolvetheproblemwhichmessageoverstockinthequeue.
5. intwaitTimes=60;
6. //for死循环不断的从队列中取出Event，并通知订阅者Subscriber执行Event
7. //Toensurethatmessagesarenotlost,enableEventHandlerwhen
8. //waitingforthefirstSubscribertoregister
9. for(;;){
10. if(shutdown||hasSubscriber()||waitTimes<=0){
11. break;
12. }
13. ThreadUtils.sleep(1000L);
14. waitTimes--;
15. }
17. for(;;){
18. if(shutdown){
19. break;
20. }
21. ////从队列取出Event
22. finalEventevent=queue.take();
23. receiveEvent(event);
24. UPDATER.compareAndSet(this,lastEventSequence,Math.max(lastEventSequence,event.sequence()));
25. }
26. }catch(Throwableex){
27. LOGGER.error("Eventlistenerexception:",ex);
28. }
29. }

这里写了两个死循环，第一个死循环可以理解为延时效果，也就是说线程启动时最大延时60秒，在这60秒中每隔1秒判断一下当前线程是否关闭，是否有订阅者，是否超过60秒。如果满足一个条件，就可以提前跳出死循环。

而第二个死循环才是真正的业务逻辑处理，会从阻塞队列中取出一个事件，然后通过receiveEvent方法进行执行。

那么，队列中的事件哪儿来的呢?此时，你可能已经想到刚才DefaultPublisher的发布事件方法被调用了。来看看它的publish方法实现：

1. @Override
2. publicbooleanpublish(Eventevent){
3. checkIsStart();
4. booleansuccess=this.queue.offer(event);
5. if(!success){
6. LOGGER.warn("Unabletopluginduetointerruption,synchronizesendingtime,event:{}",event);
7. receiveEvent(event);
8. returntrue;
9. }
10. returntrue;
11. }

可以看到，DefaultPublisher的publish方法的确就是往阻塞队列中存入事件。这里有个分支逻辑，如果存入失败，会直接调用receiveEvent，和从队列中取出事件执行的方法一样。可以理解为，如果向队列中存入失败，则立即执行，不走队列了。

最后，再来看看receiveEvent方法的实现：

1. voidreceiveEvent(Eventevent){
2. finallongcurrentEventSequence=event.sequence();
4. if(!hasSubscriber()){
5. LOGGER.warn("[NotifyCenter]the{}islost,becausethereisnosubscriber.");
6. return;
7. }
9. //通知订阅者执行Event
10. //Notificationsingleeventlistener
11. for(Subscribersubscriber:subscribers){
12. //Whethertoignoreexpirationevents
13. if(subscriber.ignoreExpireEvent()&&lastEventSequence>currentEventSequence){
14. LOGGER.debug("[NotifyCenter]the{}isunacceptabletothissubscriber,becausehadexpire",
15. event.getClass());
16. continue;
17. }
19. //BecauseunifyingsmartSubscriberandsubscriber,sohereneedtothinkofcompatibility.
20. //Removeoriginaljudgepartofcodes.
21. notifySubscriber(subscriber,event);
22. }
23. }

这里最主要的逻辑就是遍历DefaultPublisher的subscribers(订阅者集合)，然后执行通知订阅者的方法。

那么有朋友要问了这subscribers中的订阅者哪里来的呢?这个还要回到NacosNamingService的init方法中：

1. //将Subscribe注册到Publisher
2. NotifyCenter.registerSubscriber(changeNotifier);

该方法最终会调用NotifyCenter的addSubscriber方法：

1. privatestaticvoidaddSubscriber(finalSubscriberconsumer,Class<?extendsEvent>subscribeType,
2. EventPublisherFactoryfactory){
4. finalStringtopic=ClassUtils.getCanonicalName(subscribeType);
5. synchronized(NotifyCenter.class){
6. //MapUtils.computeIfAbsentisaunsafemethod.
7. MapUtil.computeIfAbsent(INSTANCE.publisherMap,topic,factory,subscribeType,ringBufferSize);
8. }
9. //获取时间对应的Publisher
10. EventPublisherpublisher=INSTANCE.publisherMap.get(topic);
11. if(publisherinstanceofShardedEventPublisher){
12. ((ShardedEventPublisher)publisher).addSubscriber(consumer,subscribeType);
13. }else{
14. //添加到subscribers集合
15. publisher.addSubscriber(consumer);
16. }
17. }

其中核心逻辑就是将订阅事件、发布者、订阅者三者进行绑定。而发布者与事件通过Map进行维护、发布者与订阅者通过关联关系进行维护。

发布者找到了，事件也有了，最后看一下notifySubscriber方法：

1. @Override
2. publicvoidnotifySubscriber(finalSubscribersubscriber,finalEventevent){
4. LOGGER.debug("[NotifyCenter]the{}willreceivedby{}",event,subscriber);
5. //执行订阅者Event
6. finalRunnablejob=()->subscriber.onEvent(event);
7. finalExecutorexecutor=subscriber.executor();
9. if(executor!=null){
10. executor.execute(job);
11. }else{
12. try{
13. job.run();
14. }catch(Throwablee){
15. LOGGER.error("Eventcallbackexception:",e);
16. }
17. }
18. }

逻辑比较简单，如果订阅者定义了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服务定义的事件机制就将这么多，下篇我们来讲讲故障转移和缓存的实现。

原文链接：https://mp.weixin.qq.com/s/RqqCZEBrpeVqMnKxnyiFhw

如果您对该产品感兴趣，请填写办理（客服微信：xiaoxiongyidong）

您的姓名 *

身份证号码 *

手机号码 *

选择地区 *

详细地址 *

留言备注 *