Spring Cloud(七):注册中心nacos-客户端视角

大家好，我是杰哥

上周，我们通过文章Spring Cloud（六）：注册中心nacos-服务端视角篇，站在nacos的服务端的视角，对nacos作为注册中心的服务端节点之间的选举、心跳、服务注册以及同步等主要动作进行了梳理说明

那么，如果想要对与nacos作为注册中心有一个全面的了解呢，今天就继续跟着杰哥一起，转向客户端的角度，来探究一波吧~

关于如何实现Nacos客户端，以及客户端的consumer如何如调用provider的服务，可以参考文章Spring Cloud（五）：注册中心-nacos篇，进行实现

等你实现以后，再进入下面的源码探究环节，可能会更快速进入节奏哦~

首先，我们从初始化步骤开始，进入源码环节

一 准备

初始工作

nacos作为注册中心，它的初始化是由NacosNamingService.init()方法进行的

我们一起看看，init()方法里都做了什么

1）初始化命名空间namespace：

initNamespaceForNaming()

在Nacos中，命名空间用于租户粗粒度隔离，同时还可以进行环境的区别，如开发环境和测试环境等等

2）初始化服务器地址：initServerAddr()

3）初始化web上下文：initWebRootContext()

它支持通过阿里云EDAS进行部署

4）初始化缓存目录：initCacheDir()

5）初始化日志文件名称：调用initLogName()方法，从配置中获取日志文件

接下来，就是我们要看的重点了

1）初始化事件分发监听器EventDispatcher

当客户端订阅了某个服务信息后，会以Listener的方式注册到EventDispatcher的队列中，当有服务变化的时候，便会通知订阅者

2）初始化服务端代理NamingProxy

用于客户端与服务端的通信

3）初始化心跳通信类BeatReactor

用于维持与服务器之间的心跳通信，上报客户端注册到服务端的服务信息

4）初始化服务信息更新器HostReactor

用于客户端服务的订阅，以及从服务端更新服务信息。对了，也就是说客户端会使用HostReactor进行订阅操作，订阅了之后，上面说到的事件分发监听器EventDispatcher会将服务信息通知给订阅者咯~

二 重点

步骤源码

01.服务更新

EventDispatcher是一个事件分发器，它维护了一个有变更的服务的队列

changedServices

private BlockingQueue<ServiceInfo> changedServices

= new LinkedBlockingQueue<ServiceInfo>();

以及一个对于某个服务的监听者列表ConcurrentMap:observerMap

private ConcurrentMap<String, List<EventListener>> observerMap
    = new ConcurrentHashMap<String, List<EventListener>>();

其中key为服务名称，value为监听这个服务变更的客户端实例列表

也就是说，EventDispatcher会实时地将服务变化信息，同步给监听该服务变化信息的客户端，使得该客户端可以进行后续动态操作

小结 1 EventDispatcher工作流程

02.心跳

首先，在初始化NacosNamingService.init()中，我们看到nacos已经对心跳类BeatReactor进行了初始化，那么在什么时候会开始进行第一次心跳呢？

答案是：实例第一次注册的时候。第一次开启并执行了心跳任务，会再开启下一次心跳的定时任务，一直继续下去，直到该实例下线

1）进入addBeatInfo()方法

实例注册的时候调用了addBeatInfo()方法，并且默认时间间隔为5s

2）进入BeatReactor.addBeatInfo()方法

该方法通过ScheduledExecutorService执行BeatTask任务

那么只需要看一下该方法的run()方法，就可以知道心跳任务的具体执行逻辑了

3）进入BeatTask.run()

该方法会调用serverProxy.sendBeat()方法，向服务端发送自身的健康状况

4）进入serverProxy.sendBeat()方法

调用reqAPI()方法，调用服务端接口：/v1/ns/instance/beat

那么，来到服务端，我们看看，接收到心跳请求之后，服务端的处理

进入服务端

服务端的/beat接口，接收到来自客户端的心跳包，调用RaftCore.receivedBeat()方法进行心跳包的处理

1）进入RaftCore#receivedBeat()方法

分为两个部分来看

第一部分 逻辑判断，排除数据过期等场景

a 分别构造当前节点信息local和发送心跳的节点信息remote

b 若发送节点不是leader，则不符合逻辑（只能由leader发起心跳），抛出异常

c 如果local的任期大于remote，说明信息已过期，抛出异常

d 若本地节点不是follower状态，不合逻辑，将本地的节点状态更新为follower

第二部分 收集数据，实现服务的更新

a 遍历请求参数中的datums，如果Follwoer不存在这个datumKey或者时间戳比较旧，则收集这个datumKey

b 当datnum的数量小于50时，继续进行收集

c 当数量大于等于50时，进行异步发送，获取对应的50个最新的Datnum对象

第三部分 将数据添加到缓存中，并更新参数

a 调用RaftStore#write()方法，将Datum序列化为json写到本地缓存中

b 将Datum存放到RaftCore的datums集合中

c 调用notifier.addTask()通知对应的RaftListener，删除key对应的旧的Datum

d 重置leaderDue时间

e 更新本地节点的任期term

小结 2 客户端心跳流程

关于服务端后续的具体处理流程，依然可以参考Spring Cloud（五）：注册中心-nacos篇哦~

03.服务更新

HostReactor用于获取、保存、更新各服务实例信息

通过维护一个服务的map集合:serviceInfoMap实现

private Map<String, ServiceInfo> serviceInfoMap;

具体的话，来一起看看~

1）获取服务端的服务信息

当客户端获取服务信息的时候，HostReactor就把服务信息保存到serviceInfoMap中，并通过UpdateTask能够周期性的从服务端获取订阅服务的最新信息，具体代码如下

我们看到，获取服务信息的方法getServiceInfo()方法中，获取到服务信息，然后将服务信息保存在serviceInfoMap中

2）更新服务信息

进入updateTask.run()方法

我们看到该方法实际上是会去定期判断当前是否有服务信息发生变化，若有变化，则会调用updateServiceNow()，来更新服务信息

3）进入updateServiceNow()方法

该方法通过服务代理调用queryList()方法，获取服务信息列表，并通过

processServiceJSON()方法，进行结果信息的处理

首先我们先来看看queryList()方法

4）进入queryList()方法

调用服务端接口/instance/list方法，获取服务信息列表

5）再来看看processServiceJSON()方法

分为两部分处理

第一部分：原有服务不为空的情况处理

第二部分：原有服务信息为空的情况处理

我们看到，总的来说，processServiceJSON()方法会将获取到的最新服务信息列表存放在serviceInfoMap中，于是服务信息便得到了更新

当然，我们也可以看到，processServiceJSON()方法更新完服务之后，也会调用eventDispatcher.serviceChanged()，将服务的变更情况，发布到队列中，也会调用DiskCache.write(),将服务信息持久化到磁盘中

同时，在这里也简单提一下，HostReactor还持有以下两个对象

pushReceiver对象，这个对象呢，用于通过UDP协议从服务器获取推送的信息，当然也会更新到serviceInfoMap当中

failoverReactor对象，这个对象呢，用于故障转移。当服务端不可用的时候，会切换到本地缓存模式，从缓存中获取服务信息。

在我看来，这一点类似于Eureka中，客户端也会将服务信息存储在本地缓存，当注册中心服务不可用时，也不会马上调用失败，就极大地保证了服务的可用性

小结3 服务信息维护角色-HostReactor的原理

好了，看完了 服务的心跳和服务更新机制，接下来，再一起进入服务上线与下线的步骤中

04.上线（注册）

说到服务的注册呢，我们先从服务Nacos的服务自动注册类说起

1) 查看NacosAutoServiceRegistration类

该类继承了AbstractAutoServiceRegistration类，而这个类，实现了ApplicationListener接口，并实现了其onApplicationEvent()方法，使得在服务启动之后，可以执行一些处理

我们看到，这里调用了bind()方法

2）进入bind()方法

在bind()方法中，调用了start()方法

3)进入start()方法

在start()方法中，便调用了register()方法，进行服务的注册操作

4）进入registrer()方法

这里使用策略模式，进入到

NacosServiceRegistry.registerInstance()方法

5）进入registerInstance()方法

6）进入实例注册方法registerInstance()

可以看到，会首先发送心跳，再使用服务代理，调用服务端注册接口，发送注册请求

7）进入NamingProxy.registerService()

首先获取实例信息，封装参数，然后调用服务端/nacos/ns/instance的post接口，进行服务的注册请求

进入服务端

1）register接口接受到客户端的请求，调用接口registerInstance()方法进行实例信息注册

2）进入ServiceManager#registerInstance()方法

a 创建空service

b 调用addInstance()方法添加新的实例

3）进入addInstance()方法

该方法主要添加 instance 到缓存中，并且持久化

4）进入addIpAddresss()方法

该方法通过调用updateIpAddresss()方法进行具体处理

5）进入updateIpAddresss()方法

我们看到该方法实际上是通过调用setValid()方法，将旧实例列表与新实例列表进行合并的

6）实例信息持久化consistencyService.put(key, instances)

再来回到实例信息持久化的方法，该方法通过调用RaftCore#signalPublish()方法，进行具体的实例信息持久化

7）实现实例信息持久化，分为两个部分

第一部分 基本判断

a 若节点不为leader状态，则转发请求给leader

b 若节点为leader状态，则将包发送给所有follower

第二部分 发送消息，同步给大多数节点

发送消息，同步等待，接受到大多数节点的响应之后，返回成功

小结 3 客户端实例注册步骤

如下图

04.下线（注销）

当服务端不可用，或者主动下关闭客户端服务实例时，会进入deregister()方法，取消服务的注册

比如当手动停掉当前的生产者服务，控制台会打出这样的日志

从日志中我们也可以看到，实际上它是调用了NacosServiceRegistry.deregister()方法，进行服务的下线操作，

1）进入NacosServiceRegistry.deregister()方法

该方法调用namingService.deregisterInstance()方法进行处理

2）进入namingService.deregisterInstance()方法

该方法会分别进行移除心跳，然后向服务端发送取消服务注册的请求

3）进入serverProxy.deregisterService()方法

该方法则使用reqAPI()调用服务端的/nacos/v1/ns/instance的DELETE接口

进入服务端

1）接口接收到客户端的取消注册请求，调用方法removeInstance()

2）进入该removeInstance()方法

3）将该方法添加到taskDispatcher中，后续将进行删除该实例信息的操作

小结 4 客户端实例取消注册步骤

三 总结

总而言之

好了，今天的推送到这里就结束啦

与上一篇一样，这篇也是花了蛮多心血在里面的，也不是说自己有什么执念，而是你们的反馈与自己的收获，让我有了坚持的动力，感谢自己，也感谢你们~

本篇文章，主要从以下几点进行介绍：

1 初始化工作介绍

2 源码探究 - 事件分发器EventDispatcher

3 源码探究 - 心跳

4 源码探究 - 服务更新HostReactor

5 源码探究 - 注册和下线

站在客户端的角度，分别探究了心跳、服务更新、服务注册和下线这几个主要事件。通过分析客户端、服务端的处理流程，以及客户端与服务端的交互，将一个动作整体贯穿起来，让大家更全面地了解到nacos注册中心的原理机制，每部分也都用流程图，呈现给了大家。

相信大家结合前两篇关于nacos作为注册中心的基本机制有了一定的了解了！

嗯，就这样。每天学习一点，时间会见证你的强大~

下期预告：

Spring Cloud(八)：注册中心-Consul