主要功能

Nacos客户端主要具有如下三个能力

1. 服务实例注册：把服务实例信息写入注册中心，可通过注册中心查找提供服务的相关实例
2. 服务实例摘除：把服务实例信息从注册中心移除
3. 服务订阅：按照服务从注册中心订阅实例信息，如果注册中心中服务实例数据发生变化，则会通知客户端进行相关变更
4. 取消订阅：取消对某个服务的订阅
   需要注意的是，Nacos中的服务注册数据被设计为五层结构，包括Namespace、Group、Service、Cluster、Instance，其中Namespace和Group两层由客户端从应用配置中读取，达到应用隔离的目的，客户端注册和订阅实例的范围在Service及以下。

客户端初始化

客户端初始化主要完成如下工作

• 初始化服务实例变更通知调度器(EventDispatcher)：启动服务实例变更通知线程，监控Nacos服务器发送到客户端的服务实例变更事件，更新客户端的服务实例数据，并维护应用定制开发的监听器，通过监听器执行相关变更
• 初始化服务器代理(ServerProxy)：维护Nacos集群服务器信息，并启动两个定时线程，一个线程查询Nacos集群中的服务器列表，如果有变化，则更新客户端数据；另一个线程不断登录Nacos服务器，保证客户端的合法性
• 初始化心跳反应器(BeatReactor)：客户端初始化时，不会启动心跳线程，客户端完成注册服务实例之后，启动一个心跳线程，不断发送心跳信息到Nacos服务端，Nacos服务端检查该实例已注册则返回成功响应信息，如果Nacos返回不成功，则会重新注册该实例，达到补偿注册实例的效果
• 初始化服务实例反应器(HostReactor)：维护服务实例信息，并启动三个失败还原相关的线程和一个服务推送信息接收线程。在三个失败还原相关线程中，第一个线程在客户端初始化时检查失败还原目录是否为空，如果为空则立即把当前服务实例信息写入磁盘，第二个线程定时刷新失败还原的开关信息；第三个线程定时把当前的服务实例信息写入磁盘；另外，服务推送信息接收的线程循环读取从Nacos服务器推送过来的信息服务实例，如果接收到了信息，就会以共享内存的形式触发服务实例变更通知器
• 获取服务器地址、服务器API路径、本地缓存文件地址、日志文件名称等
  Nacos客户端初始化代码如下，本文引用的代码都是1.2.0版本

private void init(Properties properties) {namespace = InitUtils.initNamespaceForNaming(properties);// 把构造函数的服务器地址赋值给内部对象initServerAddr(properties); // 获取服务器API路径InitUtils.initWebRootContext();// 获取本地缓存地址initCacheDir();// 获取日志文件名称initLogName(properties);// 服务实例变更通知器eventDispatcher = new EventDispatcher(); // 服务器代理serverProxy = new NamingProxy(namespace, endpoint, serverList, properties); // 心跳反应器beatReactor = new BeatReactor(serverProxy, initClientBeatThreadCount(properties)); // 实例反应器hostReactor = new HostReactor(eventDispatcher, serverProxy, cacheDir, isLoadCacheAtStart(properties),initPollingThreadCount(properties));}

重要数据结构

EventDispatcher

// EventDispatcher
public class EventDispatcher {.....// 变更的服务信息，共享内存，用于通知服务变更事件private BlockingQueue<ServiceInfo> changedServices = new LinkedBlockingQueue<ServiceInfo>();// Service对应的监听器private ConcurrentMap<String, List<EventListener>> observerMap= new ConcurrentHashMap<String, List<EventListener>>(); ......
}

如上，changedServices是一个共享内存，用于服务事件变更通知；observerMap用于存放应用定义的监听器，key是服务标识，value是监听器列表

NameProxy

// NameProxy
public class NamingProxy {.....private List<String> serverList; ......
}

如上，serverList是Nacos集群服务器列表

BeatReactor

// BeatReactor
public class BeatReactor {.....public final Map<String, BeatInfo> dom2Beat = new ConcurrentHashMap<String, BeatInfo>();......
}

如上，dom2Beat用于存放心跳信息，可以通过这些信息控制停止心跳线程，例如摘除实例时，就可以通过BeatInfo.setStopped()来停止心跳

HostReactor

// HostReactor
public class HostReactor {.....private Map<String, ServiceInfo> serviceInfoMap;......
}

如上，serviceInfoMap用于存放订阅的服务实例信息

服务实例注册

客户端完成注册服务实例时，启动一个心跳线程（AP模式下启动，CP模式不启动，CP模式由服务器主动检查客户端健康状态），该线程不断发送心跳信息到Nacos服务端，Nacos服务端检查该实例已注册之后，返回成功响应信息，如果Nacos返回不成功，则会重新注册该实例，达到补偿注册实例的效果，代码片段如下

    //NacosNamingService//添加心跳并注册服务实例public void registerInstance(String serviceName, String groupName, Instance instance) throws NacosException {if (instance.isEphemeral()) {BeatInfo beatInfo = new BeatInfo();beatInfo.setServiceName(NamingUtils.getGroupedName(serviceName, groupName));beatInfo.setIp(instance.getIp());beatInfo.setPort(instance.getPort());beatInfo.setCluster(instance.getClusterName());beatInfo.setWeight(instance.getWeight());beatInfo.setMetadata(instance.getMetadata());beatInfo.setScheduled(false);beatInfo.setPeriod(instance.getInstanceHeartBeatInterval());//添加心跳，这里会启动一个心跳线程beatReactor.addBeatInfo(NamingUtils.getGroupedName(serviceName, groupName), beatInfo);         }// 注册服务实例serverProxy.registerService(NamingUtils.getGroupedName(serviceName, groupName), groupName, instance); }

       // BeatReactor.BeatTaskpublic void run() {if (beatInfo.isStopped()) {return;}long nextTime = beatInfo.getPeriod();try {// 发送心跳JSONObject result = serverProxy.sendBeat(beatInfo, BeatReactor.this.lightBeatEnabled); long interval = result.getIntValue("clientBeatInterval");boolean lightBeatEnabled = false;if (result.containsKey(CommonParams.LIGHT_BEAT_ENABLED)) {lightBeatEnabled = result.getBooleanValue(CommonParams.LIGHT_BEAT_ENABLED);}BeatReactor.this.lightBeatEnabled = lightBeatEnabled;if (interval > 0) {nextTime = interval;}int code = NamingResponseCode.OK;if (result.containsKey(CommonParams.CODE)) {code = result.getIntValue(CommonParams.CODE);}if (code == NamingResponseCode.RESOURCE_NOT_FOUND) {//服务注册不成功时进行补偿Instance instance = new Instance();instance.setPort(beatInfo.getPort());instance.setIp(beatInfo.getIp());instance.setWeight(beatInfo.getWeight());instance.setMetadata(beatInfo.getMetadata());instance.setClusterName(beatInfo.getCluster());instance.setServiceName(beatInfo.getServiceName());instance.setInstanceId(instance.getInstanceId());instance.setEphemeral(true);try {//补偿注册实例serverProxy.registerService(beatInfo.getServiceName(),NamingUtils.getGroupName(beatInfo.getServiceName()), instance);} catch (Exception ignore) {}}} catch (NacosException ne) {NAMING_LOGGER.error("[CLIENT-BEAT] failed to send beat: {}, code: {}, msg: {}",JSON.toJSONString(beatInfo), ne.getErrCode(), ne.getErrMsg());}executorService.schedule(new BeatTask(beatInfo), nextTime, TimeUnit.MILLISECONDS);}}

服务实例摘除

通过API请求服务端摘除注册实例，并停止心跳线程（AP模式）

    // NacosNamingService// 调用API实现服务实例摘除public void deregisterInstance(String serviceName, String groupName, String ip, int port, String clusterName) throws NacosException {Instance instance = new Instance();instance.setIp(ip);instance.setPort(port);instance.setClusterName(clusterName);deregisterInstance(serviceName, groupName, instance);}

服务订阅

服务订阅完成如下几件事

• 向服务实例变更通知器(EventDispatcher)添加一个监听器，监听器保持在类EventDispatcher的成员对象observerMap里
• 调用实例反应器(HostReactor)的getServiceInfo方法查询订阅服务的实例列表，放入HostReactor内部对象serviceInfoMap中，然后再通过共享内存的形式发出服务变更事件，从而触发服务实例变更通知线程工作，调起监听器
• 向Nacos服务器写入订阅信息，以便于服务器信息发生变化时，通知本方，由服务推送信息接收线程接收处理

// EventDispatcher
public class EventDispatcher {.....// 变更的服务信息，共享内存，用于通知服务变更事件private BlockingQueue<ServiceInfo> changedServices = new LinkedBlockingQueue<ServiceInfo>();// Service对应的监听器private ConcurrentMap<String, List<EventListener>> observerMap= new ConcurrentHashMap<String, List<EventListener>>(); ......
}

调用getServiceInfo获取的服务实例信息保存在HostReactor的内部对象serviceInfoMap里

    // NacosNamingService// 服务订阅方法public void subscribe(String serviceName, String groupName, List<String> clusters, EventListener listener) throws NacosException {// 调用getServiceInfo获取服务实例// 添加监听器        eventDispatcher.addListener(hostReactor.getServiceInfo(NamingUtils.getGroupedName(serviceName, groupName),StringUtils.join(clusters, ",")), StringUtils.join(clusters, ","), listener);}

// HostReactor
public class HostReactor {......private Map<String, ServiceInfo> serviceInfoMap;......
}

在getServiceInfo中，也会放入类EventDispatcher的内部对象changedServices中，触发服务实例变更通知线程工作

取消订阅

从向服务实例变更通知器添中删除监听器

    // NacosNamingServicepublic void unsubscribe(String serviceName, String groupName, List<String> clusters, EventListener listener) throws NacosException {eventDispatcher.removeListener(NamingUtils.getGroupedName(serviceName, groupName), StringUtils.join(clusters, ","), listener);}

线程模型

服务实例变更通知线程

• JVM中名称com.alibaba.nacos.naming.client.listener
• 监控服务实例变更事件，更新客户端的服务实例数据，并维护应用定制开发的监听器，通过监听器执行相关变更
• 实现类：EventDispatcher.Notifier

Nacos服务器列表更新线程

• JVM中名称com.alibaba.nacos.client.naming.updater
• 定时从Nacos集群更新服务器列表
• 实现类：NamingProxy.匿名类

Nacos服务器登录线程

• JVM中名称com.alibaba.nacos.client.naming.updater
• 定时想Nacos服务器发送登录请求，保证客户端的合法性
• 实现类：NamingProxy.匿名类

心跳线程

• JVM中名称com.alibaba.nacos.naming.beat.sender
• 不停向服务器发送心跳信息
• 实现类：BeatReactor.BeatTask

失败恢复.开关信息刷新线程

• JVM中名称com.alibaba.nacos.naming.failover
• 定时刷新失败恢复开关，如果开关打开，则从本地读取缓存文件数据
• 实现类：FailoverReactor.SwitchRefresher

失败恢复.缓存文件写入线程

• JVM中名称com.alibaba.nacos.naming.failover
• 定时把内存中的服务实例信息写入缓存文件
• 实现类：FailoverReactor.DiskFileWriter

失败恢复.初始化缓存写入线程

• JVM中名称com.alibaba.nacos.naming.failover
• 客户端初始化时，如果本地没有缓存文件，则从内存中首次写入
• 实现类：FailoverReactor.匿名类

服务推送信息接收线程

• JVM中名称com.alibaba.nacos.naming.push.receiver
• 从服务端接收服务变更数据，则更新HostReactor中保存的服务实例信息数据，并通过服务实例变更事件触发服务实例变更通知线程
• 实现类：PushReceiver