Nacos 服务发现（订阅）源码分析（服务端）

🕗 发布于 2024-07-20 05:53 服务发现 java Nacos Alibaba Spring Cloud

前言：

前文我们分析了 Nacos 服务发现（订阅）的流程，从 Nacos Client 端的源码分析了服务发现的过程，服务发现最终还是要调用 Nacos Server 端来获取服务信息，缓存到客户端本地，并且会定时向 Nacos Server 端发送请求，获取服务信息，本篇我们从 Nacos Server 来分析一下服务订阅源码。

Nacos 系列文章传送门：

Nacos 初步认识和 Nacos 部署细节

Nacos 配置管理模型 – 命名空间（Namespace）、配置分组（Group）和配置集ID（Data ID）

Nacos 注册中心和配置中心【实战】

服务启动何时触发 Nacos 的注册流程？

Nacos Client 端服务注册流程源码分析

Nacos Server 端服务注册流程源码分析

Nacos 服务发现（订阅）源码分析（客户端）

InstanceController#list 方法源码解析

前文我们分析到服务的发现（订阅）最终会调用 Nacos Server 端的接口，而这个接口就在 InstanceController 中，根据接口路径我们找到了对应的方法也就是 InstanceController#list 方法，源码解析如下：

//com.alibaba.nacos.naming.controllers.InstanceController#list
@GetMapping("/list")
@Secured(parser = NamingResourceParser.class, action = ActionTypes.READ)
public ObjectNode list(HttpServletRequest request) throws Exception {
//获取 namespaceId
String namespaceId = WebUtils.optional(request, CommonParams.NAMESPACE_ID, Constants.DEFAULT_NAMESPACE_ID);
//获取 serviceName
String serviceName = WebUtils.required(request, CommonParams.SERVICE_NAME);
//检查格式
NamingUtils.checkServiceNameFormat(serviceName);
//agent  java、c、c++、go、nginx、dnsf
String agent = WebUtils.getUserAgent(request);
//获取集群信息
String clusters = WebUtils.optional(request, "clusters", StringUtils.EMPTY);
String clientIP = WebUtils.optional(request, "clientIP", StringUtils.EMPTY);
//获取 udp  端口
int udpPort = Integer.parseInt(WebUtils.optional(request, "udpPort", "0"));
//获取环境信息
String env = WebUtils.optional(request, "env", StringUtils.EMPTY);
boolean isCheck = Boolean.parseBoolean(WebUtils.optional(request, "isCheck", "false"));

String app = WebUtils.optional(request, "app", StringUtils.EMPTY);

String tenant = WebUtils.optional(request, "tid", StringUtils.EMPTY);

boolean healthyOnly = Boolean.parseBoolean(WebUtils.optional(request, "healthyOnly", "false"));
//获取服务列表
return doSrvIpxt(namespaceId, serviceName, agent, clusters, clientIP, udpPort, env, isCheck, app, tenant,
healthyOnly);
}

InstanceController#list 方法本身没有什么难懂逻辑，只是从 request 中获取一些属性后，继续调用了 InstanceController#doSrvIpxt 方法，我们接着往下看。

InstanceController#doSrvIpxt 方法源码解析

InstanceController#doSrvIpxt 方法的源码比较多，大概拆分一下重要步骤，做了如下事情。

根据 namespaceId, 和 serviceName 获取服务信息。
判断是有有客户端订阅了服务，如果由客户端订阅了服务，则加入到 UDP 推送列表中，也就是之前我们分析过的 Nacos Server 是如何通知 Nacos Client 服务下线。
阀值判断，通过各种判断规则得到服务列表（判断详情请看源码分析）。
封装结果集返回。

//com.alibaba.nacos.naming.controllers.InstanceController#doSrvIpxt
public ObjectNode doSrvIpxt(String namespaceId, String serviceName, String agent, String clusters, String clientIP,
int udpPort, String env, boolean isCheck, String app, String tid, boolean healthyOnly) throws Exception {
//创建客户端对象
ClientInfo clientInfo = new ClientInfo(agent);
//创建 ObjectNode
ObjectNode result = JacksonUtils.createEmptyJsonNode();
//根据  namespaceId serviceName 获取 service
Service service = serviceManager.getService(namespaceId, serviceName);
//缓存时间 默认 10 秒
long cacheMillis = switchDomain.getDefaultCacheMillis();

// now try to enable the push
try {
// udp 端口大于0 且已经开启推送  只有客户端订阅了  udp  端口才会大于0
if (udpPort > 0 && pushService.canEnablePush(agent)) {
//添加当前客户端 IP、UDP端口到 PushService 中 会作为可推送的目标客户端添加给推送服务组件
pushService
.addClient(namespaceId, serviceName, clusters, agent, new InetSocketAddress(clientIP, udpPort),
pushDataSource, tid, app);
//根据服务名 获取缓存时间默认10 秒
cacheMillis = switchDomain.getPushCacheMillis(serviceName);
}
} catch (Exception e) {
Loggers.SRV_LOG
.error("[NACOS-API] failed to added push client {}, {}:{}", clientInfo, clientIP, udpPort, e);
cacheMillis = switchDomain.getDefaultCacheMillis();
}
//service 为空判断
if (service == null) {
if (Loggers.SRV_LOG.isDebugEnabled()) {
Loggers.SRV_LOG.debug("no instance to serve for service: {}", serviceName);
}
//service 为空 构造空对象 返回
result.put("name", serviceName);
result.put("clusters", clusters);
result.put("cacheMillis", cacheMillis);
result.replace("hosts", JacksonUtils.createEmptyArrayNode());
return result;
}
//检查服务是否禁用 默认是 启用的
checkIfDisabled(service);
//服务实例 ips
List<Instance> srvedIPs;
//服务实例 ips
srvedIPs = service.srvIPs(Arrays.asList(StringUtils.split(clusters, ",")));

// filter ips using selector:
//若选择器不空 则根据选择算法选择可用的intance列表 默认情况下 选择器不做任何过滤
if (service.getSelector() != null && StringUtils.isNotBlank(clientIP)) {
srvedIPs = service.getSelector().select(clientIP, srvedIPs);
}
//serviceIps 为空判断
if (CollectionUtils.isEmpty(srvedIPs)) {
//为空
if (Loggers.SRV_LOG.isDebugEnabled()) {
Loggers.SRV_LOG.debug("no instance to serve for service: {}", serviceName);
}
//客户端类型判断
if (clientInfo.type == ClientInfo.ClientType.JAVA
&& clientInfo.version.compareTo(VersionUtil.parseVersion("1.0.0")) >= 0) {
result.put("dom", serviceName);
} else {
result.put("dom", NamingUtils.getServiceName(serviceName));
}
//构造空对象返回
result.put("name", serviceName);
result.put("cacheMillis", cacheMillis);
result.put("lastRefTime", System.currentTimeMillis());
result.put("checksum", service.getChecksum());
result.put("useSpecifiedURL", false);
result.put("clusters", clusters);
result.put("env", env);
result.set("hosts", JacksonUtils.createEmptyArrayNode());
result.set("metadata", JacksonUtils.transferToJsonNode(service.getMetadata()));
return result;
}

//存储健康和不健康的实例
//key为true的value中存放的是所有健康的instance
//key为false的value存放的是所有不健康的instance
Map<Boolean, List<Instance>> ipMap = new HashMap<>(2);
ipMap.put(Boolean.TRUE, new ArrayList<>());
ipMap.put(Boolean.FALSE, new ArrayList<>());
//服务实例遍历 区分健康和不健康的实例
for (Instance ip : srvedIPs) {
ipMap.get(ip.isHealthy()).add(ip);
}
//阀值判断 isCheck 客户端请求中如果没有传值 则默认是 fasle
if (isCheck) {
//false 标识没有达到保护阀值
result.put("reachProtectThreshold", false);
}
//获取保护阀值
double threshold = service.getProtectThreshold();
//通过健康实例除以所有实例 来判断是否触发阀值
if ((float) ipMap.get(Boolean.TRUE).size() / srvedIPs.size() <= threshold) {
//进入这里标识已经出发了保护阀值
Loggers.SRV_LOG.warn("protect threshold reached, return all ips, service: {}", serviceName);
if (isCheck) {
//启动服务保护机制
result.put("reachProtectThreshold", true);
}

//将不健康的实例全部加入到健康的实例中
//这样做的好处是可以保证服务不会那么快被打崩溃 即使有部分失败的 但是还是有可用的服务 
//不健康的实例存在的目的就是分流 缓解健康服务的压力
ipMap.get(Boolean.TRUE).addAll(ipMap.get(Boolean.FALSE));
//清空不健康的实例
ipMap.get(Boolean.FALSE).clear();
}
//阀值判断
if (isCheck) {
result.put("protectThreshold", service.getProtectThreshold());
result.put("reachLocalSiteCallThreshold", false);

return JacksonUtils.createEmptyJsonNode();
}
//能够走到这里 标识没有出发 阀值保护
ArrayNode hosts = JacksonUtils.createEmptyArrayNode();
//遍历实例
for (Map.Entry<Boolean, List<Instance>> entry : ipMap.entrySet()) {
List<Instance> ips = entry.getValue();
//如果只需要健康的实例 那就跳过不健康的实例
if (healthyOnly && !entry.getKey()) {
continue;
}
//遍历服务实例
for (Instance instance : ips) {

// remove disabled instance:
//移除禁用的实例
if (!instance.isEnabled()) {
continue;
}
//创建空对
ObjectNode ipObj = JacksonUtils.createEmptyJsonNode();
//构建实例对象
ipObj.put("ip", instance.getIp());
ipObj.put("port", instance.getPort());
// deprecated since nacos 1.0.0:
ipObj.put("valid", entry.getKey());
ipObj.put("healthy", entry.getKey());
ipObj.put("marked", instance.isMarked());
ipObj.put("instanceId", instance.getInstanceId());
ipObj.set("metadata", JacksonUtils.transferToJsonNode(instance.getMetadata()));
ipObj.put("enabled", instance.isEnabled());
ipObj.put("weight", instance.getWeight());
ipObj.put("clusterName", instance.getClusterName());
if (clientInfo.type == ClientInfo.ClientType.JAVA
&& clientInfo.version.compareTo(VersionUtil.parseVersion("1.0.0")) >= 0) {
ipObj.put("serviceName", instance.getServiceName());
} else {
ipObj.put("serviceName", NamingUtils.getServiceName(instance.getServiceName()));
}

ipObj.put("ephemeral", instance.isEphemeral());
hosts.add(ipObj);

}
}
//设置服务实例列表
result.replace("hosts", hosts);
//客户端类型判断
if (clientInfo.type == ClientInfo.ClientType.JAVA
&& clientInfo.version.compareTo(VersionUtil.parseVersion("1.0.0")) >= 0) {
result.put("dom", serviceName);
} else {
result.put("dom", NamingUtils.getServiceName(serviceName));
}
//返回结果
result.put("name", serviceName);
result.put("cacheMillis", cacheMillis);
result.put("lastRefTime", System.currentTimeMillis());
result.put("checksum", service.getChecksum());
result.put("useSpecifiedURL", false);
result.put("clusters", clusters);
result.put("env", env);
result.replace("metadata", JacksonUtils.transferToJsonNode(service.getMetadata()));
return result;
}

PushService#addClient 方法源码解析

我们上面在分析 InstanceController#doSrvIpxt 方法时候，提到如果客户端的订阅了该服务，Nacos 服务端会进行通过 UDP 推送给客户端最新的服务信息，而这个操作就是由 PushService 类实现的，PushService#addClient 方法只是把服务相关信息加入到了推送列表中。

//com.alibaba.nacos.naming.push.PushService#addClient(java.lang.String, java.lang.String, java.lang.String, java.lang.String, java.net.InetSocketAddress, com.alibaba.nacos.naming.push.DataSource, java.lang.String, java.lang.String)
public void addClient(String namespaceId, String serviceName, String clusters, String agent,
InetSocketAddress socketAddr, DataSource dataSource, String tenant, String app) {
//构造 PushClient 对象
PushClient client = new PushClient(namespaceId, serviceName, clusters, agent, socketAddr, dataSource, tenant,
app);
//加入到推送列表中
addClient(client);
}


//com.alibaba.nacos.naming.push.PushService#addClient(com.alibaba.nacos.naming.push.PushService.PushClient)
public void addClient(PushClient client) {
// client is stored by key 'serviceName' because notify event is driven by serviceName change
//获取 serviceKey
String serviceKey = UtilsAndCommons.assembleFullServiceName(client.getNamespaceId(), client.getServiceName());
//从 客户端 map  中获取 当前 client 对象
ConcurrentMap<String, PushClient> clients = clientMap.get(serviceKey);
//为空 判断
if (clients == null) {
//为空 则加入到 客户端 map 中
clientMap.putIfAbsent(serviceKey, new ConcurrentHashMap<>(1024));
//加入后获取
clients = clientMap.get(serviceKey);
}
//获取之前的 client 对象
PushClient oldClient = clients.get(client.toString());
if (oldClient != null) {
//为空空 刷新 其实是修改 client 最后一次引用时间 可以理解为更新时间
oldClient.refresh();
} else {
//为空 也就是还没有注册这个推送目标客户端  将 client 加入到 clients
PushClient res = clients.putIfAbsent(client.toString(), client);
if (res != null) {
Loggers.PUSH.warn("client: {} already associated with key {}", res.getAddrStr(), res.toString());
}
Loggers.PUSH.debug("client: {} added for serviceName: {}", client.getAddrStr(), client.getServiceName());
}
}

PushService#onApplicationEvent 方法源码分析

上面我们分析到客户端获取服务信息的时候，服务端会判断是否有客户端订阅了该服务信息，如果有，则会出发推送给客户端，最终会把服务信息封装成一个 PushClient 加入到 clientMap 中，前文我们分析了客户端是如果感知服务下线的，其中也发现了一个 clientMap 的存储结构，而在 PushService#onApplicationEvent 方法会注册一个延时任务并将该 future 放入 futureMap，该延时任务会从 clientMap 获取指定namespaceId、 serviceName 的client 集合，遍历 client 集合，判断 client 是否是 zombie（僵尸） client，如果是的则移除该 client，否则创建 Receiver.AckEntry，然后通过 UDP 的方式推送给 client，执行完毕后会从 futureMap 移除该 future，至此回到了我们前一篇分析的地方，后续就是我们熟悉的流程，不在重复分析了。

//com.alibaba.nacos.naming.push.PushService#onApplicationEvent
public void onApplicationEvent(ServiceChangeEvent event) {
//从事件对象中获取到 service
Service service = event.getService();
//获取 servicename
String serviceName = service.getName();
//获取名称空间id
String namespaceId = service.getNamespaceId();
//使用延时任务 延时1 秒 通过 UDP 的方式来发送
Future future = GlobalExecutor.scheduleUdpSender(() -> {
try {
Loggers.PUSH.info(serviceName + " is changed, add it to push queue.");
//从缓存map中获取当前服务的内层map 内层map中存放着当前服务的所有Nacos Client的
//根据 namespaceId 和 serviceName 获取对应的 client 信息
ConcurrentMap<String, PushClient> clients = clientMap
.get(UtilsAndCommons.assembleFullServiceName(namespaceId, serviceName));
//为空判断 如果为空 就没有必要推送了
if (MapUtils.isEmpty(clients)) {
return;
}
//创建缓存 map
Map<String, Object> cache = new HashMap<>(16);
//获取当前时间的 纳秒
long lastRefTime = System.nanoTime();
//遍历所有的 client 信息
for (PushClient client : clients.values()) {
//是否是僵尸客户端
if (client.zombie()) {
Loggers.PUSH.debug("client is zombie: " + client.toString());
//如果是的话 就移除僵尸客户端
clients.remove(client.toString());
Loggers.PUSH.debug("client is zombie: " + client.toString());
continue;
}
//ACK
Receiver.AckEntry ackEntry;
Loggers.PUSH.debug("push serviceName: {} to client: {}", serviceName, client.toString());
//获取推送 key
String key = getPushCacheKey(serviceName, client.getIp(), client.getAgent());
byte[] compressData = null;
Map<String, Object> data = null;
//switchDomain.getDefaultPushCacheMillis() 默认是 10秒 因此不会进入 if
if (switchDomain.getDefaultPushCacheMillis() >= 20000 && cache.containsKey(key)) {
org.javatuples.Pair pair = (org.javatuples.Pair) cache.get(key);
compressData = (byte[]) (pair.getValue0());
data = (Map<String, Object>) pair.getValue1();

Loggers.PUSH.debug("[PUSH-CACHE] cache hit: {}:{}", serviceName, client.getAddrStr());
}
//封装 ackEntry  将客户端信息封装到 ackEntry 
if (compressData != null) {
ackEntry = prepareAckEntry(client, compressData, data, lastRefTime);
} else {
//这里初始化了需要推送的 客户端
ackEntry = prepareAckEntry(client, prepareHostsData(client), lastRefTime);
if (ackEntry != null) {
cache.put(key, new org.javatuples.Pair<>(ackEntry.origin.getData(), ackEntry.data));
}
}

Loggers.PUSH.info("serviceName: {} changed, schedule push for: {}, agent: {}, key: {}",
client.getServiceName(), client.getAddrStr(), client.getAgent(),
(ackEntry == null ? null : ackEntry.key));
//通过 udp 协议向 Nacos 客户端推送数据
udpPush(ackEntry);
}
} catch (Exception e) {
Loggers.PUSH.error("[NACOS-PUSH] failed to push serviceName: {} to client, error: {}", serviceName, e);

} finally {
//移除 future
futureMap.remove(UtilsAndCommons.assembleFullServiceName(namespaceId, serviceName));
}

}, 1000, TimeUnit.MILLISECONDS);

//任务放入 futureMap  表示已经发送了 udp 到客户端的服务实例
futureMap.put(UtilsAndCommons.assembleFullServiceName(namespaceId, serviceName), future);

}

总结：服务发现（订阅）Nacos 服务端的代码还是比较简答的，而且也有一种一通百通的感觉，分析的过程中，又回到了我们前文分析的代码，一下子就知道是怎么回事了，也更加理解了 Nacos 的设计思想。

欢迎提出建议及对错误的地方指出纠正。

原文地址：https://blog.csdn.net/weixin_42118323/article/details/140401777

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：2024“钉耙编程”中国大学生算法设计超级联赛（1）
下一篇：特征交互的艺术：在sklearn中进行特征交互性建模

进阶SpringBoot之集合 Redis
测试类在属性上方添加 @Qualifier("redisTemplate") 指定 config 包下的配置类。（在跑 Redis 测试之前，需要先安装 Redis，并开启 Red
阅读更多2024-09-24
NASA：A-Train 云分级数据集(用于深度学习模型)
ATCS 是一个数据集，旨在训练深度学习模型，以便对多角度卫星图像中的云进行体积分割。该数据集包括来自 PARASOL 任务上 POLDER 传感器的多角度偏振测量的时空对齐斑块，以及来自使用 Clo
阅读更多2024-09-24
MySQL Performance Schema 详解及运行时配置优化
MySQL 的提供了非常强大的性能监控工具，而通过合理配置的运行时表，可以灵活控制采集数据的范围和深度，避免不必要的性能开销。掌握这些配置表的使用和优化技巧，能够有效提升 MySQL 数据库的运行效率
阅读更多2024-09-24
高防服务器如何抵御网络攻击？
高防服务器作为网络安全领域的重要设备，是专门为抵御各种网络攻击而设计的，主要用来抵御分布式拒绝服务攻击和CC攻击，高防服务器能够在面对大规模流量攻击时依旧保持在线服务的可用性与稳定性，下面就来介绍一下
阅读更多2024-09-24
速盾：网页游戏部署高防服务器有什么优势？
而高防服务器可以根据需求灵活扩展服务器资源，通过添加硬件设备或调整云服务器的配置，快速响应游戏的扩展需求，提高服务器的性能和稳定性。综上所述，网页游戏部署高防服务器可以带来许多优势，包括强大的抗DDo
阅读更多2024-09-24
统信服务器操作系统【qcow2 镜像空间扩容】方案
使用 qcow2 镜像安装系统，当默认安装系统存储空间不够用时，进行自定义扩容
阅读更多2024-09-24
pvesm 命令：PVE存储管理
pvesm 命令用于管理 Proxmox VE 存储。它允许你列出、创建、删除、修改存储，以及执行其他存储管理任务。支持管理：存储卷（存储池和卷组）、磁盘镜像、ISO 和容器模板等虚拟机和容器的存
阅读更多2024-09-24
操作系统week3
内存映射文件的作用：方便程序员编程，优化I/O效率多个进程可以映射同一个文件，实现共享物理内存中，一个文件对应同一份数据，当一个进程修改文件数据时，另一个进程可以立马"看到"
阅读更多2024-09-24
记一次 RabbitMQ 消费者莫名消失问题的排查
2、OOM 不一定会导致 JVM 退出，但是 SimpleMessageListenerContainer 会捕获它，然后中止当前线程，对应的队列消费者也就无了3、业务代码 catch Error 虽
阅读更多2024-09-24
爱速搭百度低代码开发平台
现在问题卡在database报错这里。
阅读更多2024-09-24

Nacos 服务发现（订阅）源码分析（服务端）

相关文章