深度解析nacos源码之注册中心（服务发现下篇）

Myth丶恋晨 2022-12-28 15:30 112阅读 0赞

> 原创不易，转载请注明出处  
> 本文基于nacos1.4.0

**深度解析nacos源码系列文章目录**

1.  《[深度解析nacos源码之注册中心（服务注册）][nacos]》
2.  《[深度解析nacos源码之注册中心（服务注册流程图补充）][nacos 1]》
3.  《[深度解析nacos源码之注册中心（服务主动下线）][nacos 2]》
4.  《[深度解析nacos源码之注册中心（服务续约与故障下线）][nacos 3]》
5.  《[深度解析nacos源码之注册中心（服务发现上篇）][nacos 4]》

### 文章目录 ###

*   *   *  前言
         *  1. 关于客户端服务订阅的补充
         *  2. 服务端异步通知
         *   *  2.1 PushService
         *  3. 客户端处理通知
         *   *  3.1 PushReceiver
             *  3.2 HostReactor\#processServiceJson
             *  3.3 EventDispatcher\#serviceChange
         *  4. nacos服务发现在dubbo中是怎样使用的
         *  总结

--------------------

### 前言 ###

在《[深度解析nacos源码之注册中心（服务发现上篇）][nacos 4]》一文中我们介绍了nacos两种服务发现的方式，一种是直接去nacos服务器上拉取实例信息，另一种是订阅拉取，就是拉取的时候，订阅这个服务，一旦你订阅的服务实例发生变化，nacos服务端就会通知你。在上篇中关于服务实例变化，nacos服务端通知客户端这块内容我们还没有介绍，本文将介绍下服务变化通知这块的代码实现。

### 1. 关于客户端服务订阅的补充 ###

在服务发现上篇中，我们只是使用了NamingService的getAllInstances 方法作为了服务发现获取服务实例列表的方式，其实nacos还支持订阅方式，通过提供订阅的服务名称与监听器（listener ）就可以了，然后订阅的服务实例信息一旦发生变化，nacos服务端就会通知你这个client，client收到通知后就会调用你这监听器，执行对应的逻辑。

NamingService naming = NamingFactory.createNamingService(properties);
    
    // 订阅orderService 服务
    naming.subscribe("orderService", new EventListener() { 
        @Override
        public void onEvent(Event event) { 
            System.out.println(((NamingEvent) event).getServiceName());
            System.out.println(((NamingEvent) event).getInstances());
        }
    });

就是上面这个样子的，直接调用subscribe 的方法就对该服务订阅了，一旦“orderService”服务实例信息发生变化，nacos服务端就会使用udp协议通知你这个client，client收到通知后，判断与本地缓存的服务实例信息有没有变化，如果有变化的话，就会调用你这个事件监听器。  
我们看下这个subscribe 方法背后的实现源码  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70]  
先是 找到hostReactor 组件获取一下这个serviceInfo信息，这块代码是不是很熟悉，我们在上篇专门解析过，就是在getAllInstances 方法中需要订阅那个代码块中  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 1]  
就是这里，这个getServiceInfo方法主要就是看看本地有没有这个serviceInfo ，如果没有的话，就创建一个，并且立马向服务端拉取这个服务实例信息，另外带着订阅。然后就是将这个订阅搞个任务出来，定时的去服务端拉拉看看，这个任务就是防止某个nacos服务端实例挂了  
导致不能及时通知了，它时不时的去服务端刷新下订阅信息，就算是某个nacos 服务挂掉了，它也可以请求其他的nacos服务继续拉取实例，同时也告诉下nacos 服务端我这个客户端还活着，维护者一个连接那种感觉。这一块详细代码我们就不看了，主要是在上篇中已经详细介绍过了。我们接下来看看EventDispatcher 是怎么回事，这个组件从名字上看是事件分发器，其实就是nacos服务端通知过来之后，client先会跟本地缓存的服务实例信息做一下比较，如果发生了变化，就会整一个服务实例变化的通知给EventDispatcher 组件，这个组件就会根据服务信息，然后找到对应的listener ，进行执行，它其实就是管着 listener 注册与调用listener 执行调用的。知道EventDispatcher组件作用之后，它的addListener方法就很简单了，就是将订阅服务信息与listener 放到一个map中。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 2]  
就是添加到了observerMap 这个map中，另外还触发了一次服务改变的事件，这个我们后面再说。

### 2. 服务端异步通知 ###

在服务注册与服务下线时候，都是先找到ConsistencyService 一致性服务将实例信息存储到对应的存储器中，我们这里主要介绍临时节点的一致性服务，也就是找到DistroConsistencyServiceImpl 这个组件，将某个service下的所有实例信息存储到了DataStore（其实就是个map）里面去了，并且会创建一个任务放到一个队列中，整完就响应给客户端了。然后DistroConsistencyServiceImpl 有个Notifier 组件会不停的从队列中获取任务，执行任务，其实就是找到对应服务的listener进行通知，说你这个服务的实例信息发生了变化，就是调用service的onChange方法进行通知，好让这个service修改下自己里面维护的信息。在修改完的时候会有这么一行代码`getPushService().serviceChanged(this);`，这行代码就是找到Push服务，然后向那些订阅了这个服务的客户端推送最新的服务实例信息。  
先来介绍下PushService 组件

#### 2.1 PushService ####

这个组件有个静态代码块，在这个代码块中初始化了一些东西我们看下  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 3]  
先是创建了一个udp的socket，启动一个线程，执行Receiver 任务，这个任务其实就是接收客户端ack的，你给客户端推送过去了最新的实例信息，然后客户端收到之后也得给你个ack。接着就是启动一个定时任务20s一次，清理下僵尸客户端，就是处理掉几秒没刷新订阅的客户端，默认是10s。  
我们看下这个serviceChange 服务改变方法  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 4]  
使用spring的事件通知发布了一个事件，PushService 这个服务其实也是一个ServiceChangeEvent 事件的观察者，实现了onApplicationEvent 方法。  
我们看下它的具体实现  
其实就是先生成一个future，然后将future放到一个任务调度线程池中执行，延迟1s，再把future放到这个futureMap中，也就是上面serviceChange方法里面那个判断future。  
看下这个future。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 5]  
先是去clientMap这个订阅客户端map中获取这个服务的一堆订阅客户端。然后遍历这堆订阅者，先从缓存中获取，默认是不启用这个缓存的，没有的话，就自己准备这个数据，放入缓存一份， 最后是调用udpPush进行推送。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 6]  
使用udpSocket进行推送，失败就重试，默认是重试1次的。  
我们再来看下Receiver 的实现，它就是接收client ack的一个任务。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 7]  
就是接收消息，进行序列化啥的。后面的一堆都是一些指标统计的东西。  
接下来我们看下客户端收到通知是怎么处理的吧。

### 3. 客户端处理通知 ###

在HostReactor组件中，有一个PushReceiver 是在HostReactor 实例化的时候创建的。先来看下这个PushReceiver组件

#### 3.1 PushReceiver ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 8]  
在实例化的时候，开启一个udpSocket，然后整了一个调度线程池，将自己作为一个任务扔进去了。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 9]  
就是等着接收请求，然后对接收过来的信息进行反序列化，如果type是dom或者是service的话就会交给hostReactor组件的processServiceJson 来处理数据。最后生成ack 返回给nacos服务端。  
我们看下这个hostReactor的processServiceJson 方法是怎样处理的

#### 3.2 HostReactor\#processServiceJson ####

关于这个方法我们不细看，它其实就是新的服务实例信息替换下本地缓存的那个服务实例信息，然后拿新通知过来的服务实例列表信息与 本地缓存的一份实例列表信息进行比较，看看有没有发生变化，如果有的话，就通知这个EventDispatcher组件。  
![在这里插入图片描述][20201219150305457.png]  
![在这里插入图片描述][20201219150319823.png]

#### 3.3 EventDispatcher\#serviceChange ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 10]  
这个就是将serviceInfo信息添加到一个队列中，然后这个EventDispatcher 组件中有个Notifier 的任务会不断从这个队列中获取serviceInfo ，通知那些订阅这个服务的listener。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 11]  
可以看到就是从这个队列中取出serviceInfo，然后从观察者map中获取对应的lisenter集合，然后遍历这个集合，进行事件的通知。

### 4. nacos服务发现在dubbo中是怎样使用的 ###

这里还要提一嘴nacos作为dubbo的注册中心是怎样服务发现的。  
其实是使用的getAllInstances 与subscribe 结合的方式进行服务发现的。我们可以稍微看下源码  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 12]  
先是使用的getAllInstances 获取这个服务的实例信息，然后通知一下订阅者，接着就是向nacos服务端订阅这个服务信息  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 13]  
生成一个eventListener ，然后调用subscribe进行订阅。

### 总结 ###

到这我们服务发现就介绍完了，本篇主要是介绍了下服务实例改变后nacos使用udp协议通知那些订阅了的客户端，客户端收到推送过来的实例信息，会改变本地的缓存信息，然后通知执行对应的listener ，它这里面大量使用了异步处理的方式，就是生成一个任务放到队列中，然后某个线程就专门从队列中获取任务，执行相应的任务处理逻辑，这是值得我们学习的地方。最后我们介绍了下nacos 服务发现在dubbo框架中的应用。

[nacos]: https://inetyoung.blog.csdn.net/article/details/111227314
[nacos 1]: https://inetyoung.blog.csdn.net/article/details/111252029
[nacos 2]: https://inetyoung.blog.csdn.net/article/details/111290929
[nacos 3]: https://inetyoung.blog.csdn.net/article/details/111299177
[nacos 4]: https://inetyoung.blog.csdn.net/article/details/111308891
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70]: /images/20221120/45dbdfecec4a4d15adff6fca6e411dfe.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 1]: /images/20221120/f995c0d0d18d435e87743a7d7eb7b06b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 2]: /images/20221120/fcdaf4e659d445149c5983f138c5d31f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 3]: /images/20221120/097b733d8d854056a09a48ca04d9222c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 4]: /images/20221120/698555bce5bf4bf0b00d065c4b56ed05.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 5]: /images/20221120/c337b0ade69e44b2b00170a3df440c27.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 6]: /images/20221120/367bb2e5fecb45e99e37143c02fe46d4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 7]: /images/20221120/c5cb640d69f54352bfcc8fa1e6f69418.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 8]: /images/20221120/cd63e836edeb44a3b73bc6064f8fb3c5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 9]: /images/20221120/d99d2b552ed947cb92d6680c1c3c5686.png
[20201219150305457.png]: /images/20221120/afb91aa11e86425ca67d113197d0307d.png
[20201219150319823.png]: /images/20221120/479b63136603459cb76f5d3f0152305c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 10]: /images/20221120/b35c145fbacb4e61a33e1879fdaf4fd1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 11]: /images/20221120/409621fc16844778beed9301eb55a7e3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 12]: /images/20221120/bd3c63878d6347a19c7ce9ba6ecac58c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 13]: /images/20221120/5be28cf741b94fa8af791011d819e4c0.png