Nacos 2.x 源码深度解析 (八)：服务订阅机制 —— 从首次订阅到gRPC双向流变更通知

《Nacos 2.x源码深度解析》专栏目录 一、架构通信篇： 《Nacos 2.x 源码深度解析 (一)：架构整体全貌 —— 核心模块划分与版本演进》 《Nacos 2.x 源码深度解析 (二)：通信协议迭代 —— HTTP长轮询到gRPC演进》 二、配置中心篇 《Nacos 2.x 源码深度解析 (三)：配置中心客户端 —— 启动加载与自动装配》 《Nacos 2.x 源码深度解析 (四)：配置中心服务端 —— 事件总线与数据持久化》 《Nacos 2.x 源码深度解析 (五)：gRPC 推送链路 —— 配置变更下发与动态刷新》 《Nacos 2.x 源码深度解析 (六)：三级缓存体系 —— 降级兜底与故障自愈机制》 三、服务注册发现篇 《Nacos 2.x 源码深度解析 (七)：服务注册流程 —— 客户端上报与服务端存储》 《Nacos 2.x 源码深度解析 (八)：服务订阅机制 —— 从首次订阅到gRPC双向流变更通知》

文章目录

• • 一、客户端查询入口：订阅模式 vs 非订阅模式
  • • 1.1 业务调用入口：从Spring Cloud到Nacos
    • 1.2 核心方法：selectInstances的分支逻辑
  • 二、首次订阅：subscribe的双重动作
  • • 2.1 订阅入口：clientProxy.subscribe()
    • 2.2 定时任务调度：scheduleUpdateIfAbsent()
    • 2.3 定时更新任务：UpdateTask.run()
    • 2.4 发起订阅请求：doSubscribe()
  • 三、服务端订阅处理：SubscribeServiceRequestHandler
  • • 3.1 服务端入口：SubscribeServiceRequestHandler.handle()
    • 3.2 服务数据聚合：ServiceStorage
  • 四、服务变更推送：从事件到通知
  • • 4.1 订阅事件触发：subscribeService()
    • 4.2 订阅事件处理：NamingSubscriberServiceV2Impl
    • 4.2、延迟任务执行引擎：PushDelayTaskExecuteEngine
    • 4.3、服务实例推送：doPushWithCallback()
  • 五、客户端接收推送：
  • • 5.1 请求接受处理：NamingPushRequestHandler.handle()
    • 5.2 服务实例处理：processServiceInfo()
    • 5.3 事件监听回调：EventListener
  • 六、全文小结

上篇深入分析了服务注册流程，其核心逻辑是把实例信息注册到ServiceManager完成落地。本篇内容将聚焦服务订阅流程，围绕如下几个核心问题展开源码剖析：

• 消费端首次订阅如何触发，怎样依托服务注册表拉取实例数据？
• 注册中心借鉴配置中心思路，依靠哪些分层机制实现故障兜底与容灾？
• 服务实例发生上下线变更后，服务端如何通过长连接向订阅客户端实时推送变动？

读者可带着上述疑问，顺着源码梳理完整订阅链路。

一、客户端查询入口：订阅模式 vs 非订阅模式

1.1 业务调用入口：从Spring Cloud到Nacos

当执行 OpenFeign 接口调用时，底层会先经过 Spring Cloud LoadBalancer 负载均衡，再通过 DiscoveryClient 接口找到 NacosDiscoveryClient 实现，最终触发 getInstances() 获取服务实例列表。NacosDiscoveryClient 在自动装配阶段由 NacosDiscoveryClientConfiguration 初始化，是服务发现接入 Nacos 的第一道入口。其内部委托 NacosServiceDiscovery.getInstances() 完成查询，该方法优先从本地 ServiceInfoHolder 缓存读取，缓存未命中时通过 gRPC 向 Nacos 服务端发起查询，服务端从内存缓存返回健康实例列表。查询结果会写入本地 ServiceCache 缓存，并将 Nacos 的 Instance 转换为 Spring Cloud 标准的 ServiceInstance 返回给调用方，完成一次完整的服务发现链路。

com.alibaba.cloud.nacos.discovery.NacosDiscoveryClient

// Nacos服务发现客户端，Spring Cloud服务发现的统一入口，通过NacosDiscoveryClientConfiguration初始化注入
public NacosDiscoveryClient(NacosServiceDiscovery nacosServiceDiscovery) {
this.serviceDiscovery = nacosServiceDiscovery;
}

// 根据服务名获取可用的服务实例列表,Spring Cloud LoadBalancer发起远程调用前触发
public List<ServiceInstance> getInstances(String serviceId) {
// 从Nacos服务端查询实例列表，内部调用NamingService.selectInstances() 获取健康实例
return Optional.of(serviceDiscovery.getInstances(serviceId))
// 缓存实例列表，ServiceCache 是客户端本地缓存，key=serviceId，value=实例列表
.map(instances -> {
ServiceCache.setInstances(serviceId, instances);
return instances;
})
.get();

——————————————————————————————————————————————————————————————————————————————

// 从Nacos服务端查询健康的服务实例列表
public List<ServiceInstance> getInstances(String serviceId) throws NacosException {
// 获取 Nacos 服务分组（默认 DEFAULT_GROUP）
String group = discoveryProperties.getGroup();

// 优先从本地ServiceInfoHolder缓存读取,缓存不存在或过期,gRPC查询服务端,服务端从内存缓存返回实例列表
List<Instance> instances = namingService().selectInstances(serviceId, group, true);

// 将 Nacos 的 Instance 列表转换为 Spring Cloud 标准的 ServiceInstance 列表
return hostToServiceInstanceList(instances, serviceId);
}

1.2 核心方法：selectInstances的分支逻辑

NacosNamingService.selectInstances() 是服务实例查询的核心入口，内部通过 getServiceInfo() 实现三级降级策略：优先读取本地 Failover 容灾文件，用于服务端完全不可达时的兜底；其次走订阅缓存模式，首次访问时通过 clientProxy.subscribe() 从服务端拉取并建立 gRPC 订阅，后续直接从本地 ServiceInfoHolder 缓存读取，同时接收服务端推送的变更通知；若指定非订阅模式，则通过 clientProxy.queryInstancesOfService() 直连服务端一次性查询，不缓存也不接收后续推送。查询到 ServiceInfo 后，再按健康状态等条件过滤，返回最终的实例列表。这套机制兼顾了实时性、容灾降级与资源开销，是 Nacos 服务发现高可用的核心保障。

com.alibaba.nacos.client.naming.NacosNamingService

// 从Nacos服务端查询服务实例列表，按条件过滤后返回
public List<Instance> selectInstances(String serviceName, String groupName, List<String> clusters, boolean healthy,
boolean subscribe) throws NacosException {
// 获取服务实例信息（走缓存或直连）
ServiceInfo serviceInfo = getServiceInfo(serviceName, groupName, clusters, subscribe);
// 按条件过滤并返回
return selectInstances(serviceInfo, healthy);
}

——————————————————————————————————————————————————————————————————————————————

// 获取服务实例信息，三级降级策略，Failover容灾文件 —> 订阅缓存（默认） —> 直连服务端查询
private ServiceInfo getServiceInfo(String serviceName, String groupName, List<String> clusters, boolean subscribe)
throws NacosException {
ServiceInfo serviceInfo;
// 将集群列表拼接为字符串（如 "DEFAULT,canary"），作为缓存键的一部分
String clusterString = StringUtils.join(clusters, ",");

// serviceInfoHolder.isFailoverSwitch() 返回 true 表示用户手动开启了容灾模式，容灾文件由运维手动放置，优先级最高，用于服务端完全不可达时的兜底
if (serviceInfoHolder.isFailoverSwitch()) {
serviceInfo = getServiceInfoByFailover(serviceName, groupName, clusterString);
// 容灾文件存在且有有效实例，直接返回，不再访问服务端
if (serviceInfo != null && serviceInfo.getHosts().size() > 0) {
return serviceInfo;
}
}

// 第二级/第三级：订阅缓存或直连查询
// subscribe=true，优先从ServiceInfoHolder缓存取，首次访问缓存为空，从clientProxy.subscribe() 拉取并订阅
// subscribe=false，clientProxy.queryInstancesOfService() 直连服务端一次性查询
serviceInfo = getServiceInfoBySubscribe(serviceName, groupName, clusterString, subscribe);
return serviceInfo;
}

——————————————————————————————————————————————————————————————————————————————

// 通过订阅或直连两种方式获取服务实例信息
private ServiceInfo getServiceInfoBySubscribe(String serviceName, String groupName, String clusterString,
boolean subscribe) throws NacosException {
ServiceInfo serviceInfo;

if (subscribe) {
// 订阅模式
serviceInfo = serviceInfoHolder.getServiceInfo(serviceName, groupName, clusterString);
// 首次访问该服务，缓存中没有任何数据 或 缓存中有旧数据但当前连接未订阅（可能发生在断线重连后，旧缓存还在但订阅关系已丢失）
if (null == serviceInfo || !clientProxy.isSubscribed(serviceName, groupName, clusterString)) {
// 委托给clientProxy.subscribe() 执行订阅逻辑
serviceInfo = clientProxy.subscribe(serviceName, groupName, clusterString);
}
} else {
// 非订阅模式（一次性查询），不订阅、不缓存、不接收后续推送
serviceInfo = clientProxy.queryInstancesOfService(serviceName, groupName, clusterString, false);
}

return serviceInfo;
}

二、首次订阅：subscribe的双重动作

以OpenFeign为例，其发起远程调用时，经由Spring Cloud LoadBalancer触发服务实例查询，基于DiscoveryClient进入 Nacos 服务发现链路。

首次调用触发懒订阅逻辑：NacosNamingService校验ServiceInfoHolder本地缓存为空，随即向 Nacos Server 发起 gRPC 订阅请求，服务端返回全量实例数据，客户端刷新本地缓存后逐层回传实例列表，负载均衡筛选节点完成对 Provider 的 HTTP 调用；后续服务实例发生上下线变更时，Nacos Server 通过 gRPC 长连接向客户端推送增量变更，自动更新ServiceInfoHolder缓存。

同服务后续再次发起 Feign 调用，无需重复请求服务端，selectInstances直接读取本地缓存返回实例，依靠首次订阅全量拉取、变更实时推送、后续本地读缓存的推拉模型，兼顾调用性能与实例数据实时性。

2.1 订阅入口：clientProxy.subscribe()

subscribe() 是服务消费方获取实例列表并建立长连接推送通道的核心入口，位于客户端代理委托层，负责协调 gRPC 代理与本地缓存的交互。方法首先将 serviceName 和 groupName 拼接为 groupName@@serviceName 格式的分组服务名，再与 clusters 拼接为完整缓存键。随后调用 scheduleUpdateIfAbsent() 启动定时更新任务作为兜底保障，即使 gRPC 推送通知全部丢失，定时任务也会在约 6 秒间隔后主动拉取最新实例列表，连续失败时退避至最长 60 秒。

获取实例数据时采用"缓存优先"策略：先从 ServiceInfoHolder 本地缓存读取，若缓存为空（首次访问）或订阅关系因断线重连已丢失，则通过 grpcClientProxy.subscribe() 向服务端发起 gRPC 订阅请求，服务端返回当前完整的 ServiceInfo 并将该连接加入推送列表。最后调用 processServiceInfo() 按 lastRefTime 时间戳对比后更新本地缓存，防止推送乱序导致数据回退。

com.alibaba.nacos.client.naming.remote.NamingClientProxyDelegate

// 订阅指定服务，获取其实例列表并建立长连接推送通道
public ServiceInfo subscribe(String serviceName, String groupName, String clusters) throws NacosException {
// 拼接服务名和分组,生成服务标识键，"DEFAULT_GROUP@@order-service"
String serviceNameWithGroup = NamingUtils.getGroupedName(serviceName, groupName);s
// 拼接完整缓存键, serviceNameWithGroup + "@@" + clusters
String serviceKey = ServiceInfo.getKey(serviceNameWithGroup, clusters);

// 启动定时更新任务
serviceInfoUpdateService.scheduleUpdateIfAbsent(serviceName, groupName, clusters);

// 优先读取本地缓存，避免不必要的网络请求
ServiceInfo result = serviceInfoHolder.getServiceInfoMap().get(serviceKey);

// 缓存为空或订阅关系丢失，向服务端发起gRPC订阅。
// result == null：首次访问该服务，缓存中没有任何数据
// !isSubscribed()：缓存中有旧数据，但订阅关系因断线重连已丢失
if (null == result || !isSubscribed(serviceName, groupName, clusters)) {
// 通过 gRPC 代理向服务端发起订阅请求
result = grpcClientProxy.subscribe(serviceName, groupName, clusters);
}

// 将最新的 ServiceInfo 更新到本地缓存
serviceInfoHolder.processServiceInfo(result);

// 返回最新的服务实例信息
return result;
}

2.2 定时任务调度：scheduleUpdateIfAbsent()

scheduleUpdateIfAbsent()是服务发现客户端为每个订阅服务启动定时兜底任务的入口，采用双重检查锁定模式保证同一服务只创建一个UpdateTask。方法先检查asyncQuerySubscribeService开关，再以groupName@@serviceName@@clusters为键查询futureMap——第一次无锁快速判断是否已存在定时任务，若不存在则加锁后再次检查，最后通过addTask()创建UpdateTask并提交到ScheduledExecutorService，同时将返回的ScheduledFuture句柄缓存到futureMap中。

该定时任务是gRPC实时推送的兜底保障：正常路径下推送通道保证秒级变更感知，一旦推送通知全部丢失，UpdateTask每隔约6秒主动向服务端查询最新实例列表并更新本地缓存。任务执行时通过lastRefTime判断期间是否已收到推送，若已推送则跳过本次查询避免重复请求；连续失败时采用指数退避延迟，失败次数越多等待越久，上限60秒基础延迟。

com.alibaba.nacos.client.naming.core.ServiceInfoUpdateService

// 为指定服务启动定时更新任务
public void scheduleUpdateIfAbsent(String serviceName, String groupName, String clusters) {
// asyncQuerySubscribeService 默认为 true，可通过系统属性关闭，如果关闭，则不启动定时更新任务，完全依赖 gRPC 推送
if (!asyncQuerySubscribeService) {
return;
}

String serviceKey = ServiceInfo.getKey(
NamingUtils.getGroupedName(serviceName, groupName), clusters);

// 如果该服务已有定时任务在运行，直接返回，无需重复创建
if (futureMap.get(serviceKey) != null) {
return;
}

// synchronized 保证原子性：防止并发场景下多个线程同时通过第一次检查
synchronized (futureMap) {
// 第二次检查：再次确认是否有其他线程已经抢先创建了任务
if (futureMap.get(serviceKey) != null) {
return;
}

// 创建并启动定时更新任务
ScheduledFuture<?> future = addTask(new UpdateTask(serviceName, groupName, clusters));

// 缓存定时任务句柄，后续调用本方法时通过第一次检查快速判断是否已存在，或者服务取消订阅时，通过 Future.cancel() 停止定时任务
futureMap.put(serviceKey, future);
}
}

2.3 定时更新任务：UpdateTask.run()

UpdateTask是gRPC实时推送的定时兜底任务，默认间隔约6秒（cacheMillis×6），连续失败时指数退避至最长60秒。每次执行先检查订阅是否已取消，已取消则自停。然后从本地缓存读取ServiceInfo：缓存为空时直接向服务端拉取全量数据并写入缓存，同时记录lastRefTime作为后续推送更新的比对标识；缓存已存在时通过比对本轮lastRefTime与服务端推送后的lastRefTime来判断期间是否有推送到达，若有则跳过本轮拉取，避免与推送重复请求，若无则主动拉取更新。拉取到的实例列表为空时递增失败计数并提前返回。最终按delayTime×2^failCount计算下一次执行的延迟，上限60秒。

com.alibaba.nacos.client.naming.core.ServiceInfoUpdateService.UpdateTask

// 定时更新任务，主动向服务端查询最新实例列表
public void run() {
long delayTime = DEFAULT_DELAY;

try {
// 检查任务是否需要停止，当该服务已取消订阅并且futureMap中已移除该任务时停止
if (!changeNotifier.isSubscribed(groupName, serviceName) && !futureMap.containsKey(serviceKey)) {
isCancel = true;
return;
}

// 从本地缓存获取当前 ServiceInfo
ServiceInfo serviceObj = serviceInfoHolder.getServiceInfoMap().get(serviceKey);

if (serviceObj == null) {
// 本地缓存为空，首次执行或缓存被清理
// 主动向服务端查询实例列表
serviceObj = namingClientProxy.queryInstancesOfService(serviceName, groupName, clusters, false);
// 更新本地缓存（触发实例变更事件和快照）
serviceInfoHolder.processServiceInfo(serviceObj);
// 从响应中获取服务端建议的缓存时间 × 6，作为下次刷新间隔
delayTime = serviceObj.getCacheMillis() * DEFAULT_UPDATE_CACHE_TIME_MULTIPLE;
// 记录本次拉取的时间戳，用于下轮判断是否有推送更新
lastRefTime = serviceObj.getLastRefTime();
return;
}

// 缓存已存在，判断期间是否有推送更新，当前缓存的lastRefTime未超过上次记录值，说明期间无推送更新
if (serviceObj.getLastRefTime() <= lastRefTime) {
// 主动拉取最新实例列表
serviceObj = namingClientProxy.queryInstancesOfService(serviceName, groupName, clusters, false);
// 更新本地缓存
serviceInfoHolder.processServiceInfo(serviceObj);
}

// 记录最新时间戳，用于下次判断
lastRefTime = serviceObj.getLastRefTime();

// 计算下一次调度间隔
delayTime = serviceObj.getCacheMillis() * DEFAULT_UPDATE_CACHE_TIME_MULTIPLE;
// 重置失败计数
resetFailCount();

} finally {
// 调度下一次执行
if (!isCancel) {
// 延迟 = delayTime × 2^failCount，上限 60 秒
executor.schedule(this, Math.min(delayTime << failCount, DEFAULT_DELAY * 60),
TimeUnit.MILLISECONDS);
}
}
}

2.4 发起订阅请求：doSubscribe()

doSubscribe() 是gRPC订阅请求的实际执行者。它构造 SubscribeServiceRequest（subscribe=true），通过 requestToServer() 发送gRPC Unary一元调用，同步等待服务端返回当前全量实例数据；成功后调用 redoService.subscriberRegistered() 将对应订阅标记为"已注册"，断线重连后 RedoScheduledTask 不会再为此订阅补发请求。它与 subscribe() 入口构成"先缓存、后发送、成功确认"的两阶段模式，即使请求成功但响应丢失，redoService中仍保留该订阅为待确认状态，后续由定时任务自动补发。

com.alibaba.nacos.client.naming.remote.gprc.NamingGrpcClientProxy

public ServiceInfo subscribe(String serviceName, String groupName, String clusters) throws NacosException {
// 先存入redoService缓存，isRegistered=false（待确认状态），断线后定时任务会扫描到此订阅，自动补发
redoService.cacheSubscriberForRedo(serviceName, groupName, clusters);
// 发送订阅请求，doSubscribe 成功后调用subscriberRegistered() 标记为已确认
return doSubscribe(serviceName, groupName, clusters);
}

——————————————————————————————————————————————————————————————————————————————

// 执行订阅操作，通过gRPC Unary向服务端发起订阅请求，并返回当前全量实例。
public ServiceInfo doSubscribe(String serviceName, String groupName, String clusters) throws NacosException {

SubscribeServiceRequest request = new SubscribeServiceRequest(
namespaceId, groupName, serviceName, clusters, true);

// 发送gRPC Unary一元调用请求，同步等待响应
SubscribeServiceResponse response = requestToServer(request, SubscribeServiceResponse.class);

// redoService在subscribe() 入口缓存了该订阅（cacheSubscriberForRedo）
// 此处标记 isRegistered=true，断线重连后定时任务不会再重试
redoService.subscriberRegistered(serviceName, groupName, clusters);

// 返回服务端返回的当前全量实例数据
return response.getServiceInfo();
}

三、服务端订阅处理：SubscribeServiceRequestHandler

3.1 服务端入口：SubscribeServiceRequestHandler.handle()

SubscribeServiceRequestHandler.handle() 是服务端处理订阅请求的入口。它先解析请求中的 namespace、serviceName、groupName，再从 ServiceStorage 中聚合该服务所有注册实例。根据请求的 subscribe 标记分流：true 时调用 clientOperationService.subscribeService() 将订阅关系写入 Client 对象的 subscribers 并发布 ClientSubscribeServiceEvent 触发 ClientServiceIndexesManager 更新索引，false 时解除订阅。无论订阅还是取消，都返回同一份 `ServiceInfo，客户端首次订阅时借此拿到全量实例，取消订阅时借此保证即使取消也至少获取一次最新数据。

com.alibaba.nacos.naming.remote.rpc.handler.SubscribeServiceRequestHandler

public SubscribeServiceResponse handle(SubscribeServiceRequest request, RequestMeta meta) throws NacosException {

String namespaceId = request.getNamespace();
String serviceName = request.getServiceName();
String groupName = request.getGroupName();

// 从请求上下文中获取应用名
String app = RequestContextHolder.getContext().getBasicContext().getApp();

// 构建服务与订阅者对象
String groupedServiceName = NamingUtils.getGroupedName(serviceName, groupName); // group@@service
Service service = Service.newService(namespaceId, groupName, serviceName, true); // ephemeral=true

// 构造订阅者信息：IP、客户端版本、应用名、集群名
Subscriber subscriber = new Subscriber(meta.getClientIp(), meta.getClientVersion(), app, meta.getClientIp(),
namespaceId, groupedServiceName, 0, request.getClusters());

// 从ServiceStorage获取实例数据，过滤后返回
ServiceInfo serviceInfo = ServiceUtil.selectInstancesWithHealthyProtection(
serviceStorage.getData(service),
metadataManager.getServiceMetadata(service).orElse(null),
subscriber.getCluster(), false, true, subscriber.getIp());

// 根据subscribe标记分流
if (request.isSubscribe()) {
// 建立订阅关系：将 subscriber 存入 Client.subscribers，并发布ClientSubscribeServiceEvent
clientOperationService.subscribeService(service, subscriber, meta.getConnectionId());
// 发布订阅追踪事件，用于监控、审计
NotifyCenter.publishEvent(new SubscribeServiceTraceEvent(...));
} else {
// 取消订阅：从Client.subscribers中移除
clientOperationService.unsubscribeService(service, subscriber, meta.getConnectionId());
NotifyCenter.publishEvent(new UnsubscribeServiceTraceEvent(...));
}

// 无论订阅还是取消，都返回当前全量实例数据
return new SubscribeServiceResponse(ResponseCode.SUCCESS.getCode(), "success", serviceInfo);
}

3.2 服务数据聚合：ServiceStorage

ServiceStorage 是Nacos实例数据的聚合与缓存层。由于实例数据分散存储在 Client 对象的 publishers 中，ServiceStorage 负责将三者串联起来——通过 ClientServiceIndexesManager 获取注册了该服务的所有 clientId，通过 ClientManager 获取每个 Client 并取出其 InstancePublishInfo，再通过 InstanceUtil 转为 API 层的 Instance 对象，最终聚合为完整的 ServiceInfo 返回。内部维护 serviceDataIndexes 缓存避免重复聚合，getPushData() 在缓存未命中时重建，removeData() 在实例变更时令缓存失效。与 ServiceManager 的分工明确：后者只管理 Service 的元数据和唯一性，前者负责实例数据的组装和输出

（数据模型请参阅 《Nacos 2.x 源码深度解析 (七)：服务注册流程 —— 客户端上报与服务端存储》的第四节内容"核心结构：服务注册表设计与存储"）

com.alibaba.nacos.naming.core.v2.index.ServiceStorage

// 客户端-服务索引管理器，提供 Service → clientId 的倒排索引
private final ClientServiceIndexesManager serviceIndexesManager;

// 客户端管理器，通过clientId获取Client对象（存储实例数据）
private final ClientManager clientManager;

// 元数据管理器，用于获取实例的扩展元数据
private final NamingMetadataManager metadataManager;

// Service —> 聚合后的 ServiceInfo 缓存
private final ConcurrentMap<Service, ServiceInfo> serviceDataIndexes;

// Service —> 该服务涉及的集群名集合
private final ConcurrentMap<Service, Set<String>> serviceClusterIndex;

// 获取服务的实例数据。缓存命中直接返回，未命中时通过getPushData() 从索引重建。
public ServiceInfo getData(Service service) {
return serviceDataIndexes.containsKey(service)
? serviceDataIndexes.get(service)
: getPushData(service);
}

——————————————————————————————————————————————————————————————————————————————

// 从索引重建 ServiceInfo（缓存未命中时触发）
public ServiceInfo getPushData(Service service) {
ServiceInfo result = emptyServiceInfo(service);

// 获取全局唯一的Service实例（作为缓存 key）
Service singleton = ServiceManager.getInstance().getSingleton(service);

// 从索引中聚合所有实例
result.setHosts(getAllInstancesFromIndex(singleton));

// 写入本地缓存
serviceDataIndexes.put(singleton, result);
return result;
}

——————————————————————————————————————————————————————————————————————————————

// 遍历所有注册了该服务的clientId，从Client中提取实例数据并聚合
private List<Instance> getAllInstancesFromIndex(Service service) {
Set<Instance> result = new HashSet<>();
Set<String> clusters = new HashSet<>();

// 遍历所有注册了此服务的 clientId
for (String each : serviceIndexesManager.getAllClientsRegisteredService(service)) {

// 通过 clientManager 获取 Client，再取实例数据
Optional<InstancePublishInfo> instancePublishInfo = getInstanceInfo(each, service);
if (instancePublishInfo.isPresent()) {
InstancePublishInfo publishInfo = instancePublishInfo.get();

if (publishInfo instanceof BatchInstancePublishInfo) {
// 批量实例：一个批量端口对应多个实例
BatchInstancePublishInfo batch = (BatchInstancePublishInfo) publishInfo;
List<Instance> batchInstance = parseBatchInstance(service, batch, clusters);
result.addAll(batchInstance);
} else {
// 普通实例：一个连接对应一个实例
Instance instance = parseInstance(service, instancePublishInfo.get());
result.add(instance);
clusters.add(instance.getClusterName());
}
}
}

// 缓存该服务的集群信息
serviceClusterIndex.put(service, clusters);
return new LinkedList<>(result);
}

——————————————————————————————————————————————————————————————————————————————

// 通过clientId从ClientManager获取Client，再取出指定服务的实例数据。
private Optional<InstancePublishInfo> getInstanceInfo(String clientId, Service service) {
Client client = clientManager.getClient(clientId);
if (null == client) {
return Optional.empty(); // 连接已断开，Client 已释放
}
return Optional.ofNullable(client.getInstancePublishInfo(service));
}

四、服务变更推送：从事件到通知

4.1 订阅事件触发：subscribeService()

当服务端接收到订阅请求时，由SubscribeServiceRequestHandler处理，通过clientOperationService.subscribeService()方法，流转到subscribeService()。该方法将客户端的订阅关系写入服务端存储。它先通过 ServiceManager 获取全局唯一的 Service 实例，再从 ClientManager 中取出对应的 Client 对象，经合法性校验后将 Subscriber 存入 Client 的 subscribers 映射，更新最后活跃时间，最后发布 ClientSubscribeServiceEvent。此事件由 ClientServiceIndexesManager 消费：更新 subscriberIndexes[service] 增加该 clientId，触发 ServiceSubscribedEvent 并驱动 NamingSubscriberServiceV2Impl 向该客户端定向推送当前全量实例数据。整个方法实现了服务端侧订阅关系的持久化与索引更新，是客户端 doSubscribe() gRPC 请求在服务端的处理终点。

com.alibaba.nacos.naming.core.v2.service.impl.EphemeralClientOperationServiceImpl

@Override
public void subscribeService(Service service, Subscriber subscriber, String clientId) {

// getSingletonIfExist() 尝试从ServiceManager中查找已注册的服务单例；如果不存在（首次订阅时服务可能尚未注册），则直接用传入的service对象
Service singleton = ServiceManager.getInstance().getSingletonIfExist(service).orElse(service);

// 通过clientId从ClientManager中取出对应的Client对象
Client client = clientManager.getClient(clientId);

// 将订阅关系写入 Client
// 存入AbstractClient.subscribers: ConcurrentHashMap<Service, Subscriber>
// key = Service, value = Subscriber（含 IP、版本、集群等信息）
client.addServiceSubscriber(singleton, subscriber);

// 发布订阅事件
NotifyCenter.publishEvent(new ClientOperationEvent.ClientSubscribeServiceEvent(singleton, clientId));
}

4.2 订阅事件处理：NamingSubscriberServiceV2Impl

当客户端首次订阅某个服务时，服务端会立即向该客户端推送一次当前完整的实例列表，无需等待下一次实例变更。这一机制由两个组件协作完成：ClientServiceIndexesManager.addSubscriberIndexes() 先将客户端 ID 加入该服务的订阅者索引，若为首次订阅则发布 ServiceSubscribedEvent 事件；NamingSubscriberServiceV2Impl.onEvent() 收到事件后将推送任务加入延迟队列，默认延迟 500ms 后向该客户端推送当前完整的实例列表。

复用延迟队列而非直接推送的设计，利用了其 500ms 合并窗口，如果订阅生效的500ms内恰好有该服务的实例发生变更（ServiceChangedEvent 触发），PushDelayTask.merge() 会将两次通知合并为一次推送，避免短时间内向新订阅者发送多条重复消息。这种"订阅即推送 + 合并去重"的设计，既保证了新订阅者能立即获取实例列表，又避免了重复通知带来的网络开销。

com.alibaba.nacos.naming.core.v2.index.ClientServiceIndexesManager

private void addSubscriberIndexes(Service service, String clientId) {
Set<String> clientIds = subscriberIndexes.computeIfAbsent(service, key -> new ConcurrentHashSet<>());

if (clientIds.add(clientId)) {
// 发送订阅事件
NotifyCenter.publishEvent(new ServiceEvent.ServiceSubscribedEvent(service, clientId));
}
}

——————————————————————————————————————————————————————————————————————————————

com.alibaba.nacos.naming.push.v2.NamingSubscriberServiceV2Impl

// 收到服务订阅事件时的回调
public void onEvent(Event event) {
ServiceEvent.ServiceSubscribedEvent subscribedEvent = (ServiceEvent.ServiceSubscribedEvent) event;
Service service = subscribedEvent.getService();
// 将推送任务加入延迟队列,客户端会在延迟到期后立即收到当前完整的实例列表
delayTaskEngine.addTask(service,
new PushDelayTask(service, PushConfig.getInstance().getPushTaskDelay(), subscribedEvent.getClientId()));
}

4.2、延迟任务执行引擎：PushDelayTaskExecuteEngine

我们在 《Nacos 2.x 源码深度解析 (七)：服务注册流程 —— 客户端上报与服务端存储》的第五节内容 “事件通知：变更信息推送及集群数据同步” 初次见过PushDelayTaskExecuteEngine推送延迟引擎，该类继承NacosDelayTaskExecuteEngine。

当NamingSubscriberServiceV2Impl监听到ServiceSubscribedEvent或ServiceChangedEvent事件时，会向引擎塞入PushDelayTask，默认延迟 500ms 后交由PushDelayTaskProcessor处理；再经由NamingExecuteTaskDispatcher按服务隔离线程分发，执行PushExecuteTask。

任务内部依托ServiceStorage组装最新实例列表，结合订阅索引筛选推送对象，最终由PushExecutor通过 gRPC 下发NotifySubscriberRequest；推送失败则依靠ServicePushCallback重新入队重试。依托任务合并特性，500ms 窗口期内同一服务多次实例变更会被合并为单次推送，在推送时效性与网络开销之间实现平衡。

com.alibaba.nacos.naming.push.v2.task.PushDelayTaskExecuteEngine

// 服务推送延迟任务执行引擎。
public class PushDelayTaskExecuteEngine extends NacosDelayTaskExecuteEngine {

protected void processTasks() {
if (!switchDomain.isPushEnabled()) {
return; // 推送已关闭，跳过
}
super.processTasks(); // 调用父类处理到期任务
}

——————————————————————————————————————————————————————————————————————————————

// 任务到期后由父类NacosDelayTaskExecuteEngine调用process()，内部通过NamingExecuteTaskDispatcher分发到对应的线程池执行，避免不同服务的推送任务互相阻塞。
private static class PushDelayTaskProcessor implements NacosTaskProcessor {

private final PushDelayTaskExecuteEngine executeEngine;

@Override
public boolean process(NacosTask task) {
PushDelayTask pushDelayTask = (PushDelayTask) task;
Service service = pushDelayTask.getService();

// 按service分派到独立线程执行，同一服务的推送串行化
NamingExecuteTaskDispatcher.getInstance()
.dispatchAndExecuteTask(service,
new PushExecuteTask(service, executeEngine, pushDelayTask));
return true;
}
}
}

——————————————————————————————————————————————————————————————————————————————

com.alibaba.nacos.naming.push.v2.task.PushExecuteTask

public void run() {
// 从ServiceStorage获取最新实例数据
PushDataWrapper wrapper = generatePushData();
// ② 确定推送目标
// pushToAll = true → indexesManager.getAllClientsSubscribeService(service)
// pushToAll = false → delayTask.getTargetClients()（仅单个客户端）
// ③ 遍历每个订阅者，通过 PushExecutor 推送
for (String each : getTargetClientIds()) {
Client client = clientManager.getClient(each);
Subscriber subscriber = client.getSubscriber(service);
delayTaskEngine.getPushExecutor().doPushWithCallback(each, subscriber, wrapper,
new ServicePushCallback(each, subscriber, wrapper.getOriginalData(), delayTask.isPushToAll()));
}
}

4.3、服务实例推送：doPushWithCallback()

pushWithCallback() 通过 connectionId 查找客户端 gRPC 连接，若连接存在则通过 connection.asyncRequest() 经双向流发送请求，并内置回调适配层将客户端的异步回执转为 onSuccess / onFail 回调通知调用方；连接已关闭时自动从 ConnectionManager 注销并视为推送成功，连接已不存在时直接跳过。调用方（如 RpcConfigChangeNotifier 和 PushExecutorRpcImpl）据此决定是否需要重试或清理本地状态。

com.alibaba.nacos.core.remote.RpcPushService

// 通过gRPC双向流向客户端推送请求，并异步等待客户端回执。
public void pushWithCallback(String connectionId, ServerRequest request,
PushCallBack requestCallBack, Executor executor) {

// 按connectionId查找客户端连接
Connection connection = connectionManager.getConnection(connectionId);

if (connection != null) {

try {
// 通过双向流异步发送
connection.asyncRequest(request, new AbstractRequestCallBack(requestCallBack.getTimeout()) {

public Executor getExecutor() { return executor; }

public void onResponse(Response response) {
// 客户端回执，按结果调用成功/失败回调
if (response.isSuccess()) {
requestCallBack.onSuccess();
} else {
requestCallBack.onFail(new NacosException(...));
}
}

public void onException(Throwable e) {
requestCallBack.onFail(e); // 网络异常
}
});

} catch (ConnectionAlreadyClosedException e) {
// 连接已关闭，注销连接，视为推送成功
connectionManager.unregister(connectionId);
requestCallBack.onSuccess();

} catch (Exception e) {
requestCallBack.onFail(e); // 其他异常，触发调用方重试
}

} else {
// 连接已不存在，无需推送，视为成功
requestCallBack.onSuccess();
}
}

五、客户端接收推送：

5.1 请求接受处理：NamingPushRequestHandler.handle()

requestReply()是客户端接收命名服务推送的入口。收到 NotifySubscriberRequest 后，提取其中的 ServiceInfo 并调用 serviceInfoHolder.processServiceInfo() 写入本地缓存；该方法内部会通过时间戳对比和实例差异计算，若数据有变更则发布 InstancesChangeEvent，由 InstancesChangeNotifier 分发到用户注册的 EventListener。

com.alibaba.nacos.client.naming.remote.gprc.NamingPushRequestHandler

public Response requestReply(Request request, Connection connection) {
NotifySubscriberRequest notifyRequest = (NotifySubscriberRequest) request;
// 委托ServiceInfoHolder处理推送的serviceInfo
serviceInfoHolder.processServiceInfo(notifyRequest.getServiceInfo());
return new NotifySubscriberResponse();
}

5.2 服务实例处理：processServiceInfo()

processServiceInfo() 是客户端处理服务端推送或查询返回的 ServiceInfo 的核心方法。它先从 serviceInfoMap 中取出旧的缓存数据，触发空推送保护检查，若服务端推了空列表且启用了空保护则跳过更新；通过校验后写入新数据覆盖旧缓存，然后通过 InstancesDiffer 计算新旧实例的差异（新增、移除、变更）；如果存在差异且未处于 failover 模式，则发布 InstancesChangeEvent 供 InstancesChangeNotifier 分发到用户的 EventListener，同时将数据写入本地磁盘快照以实现缓存持久化。该方法同时服务于推送通道（NamingPushRequestHandler）和定时拉取通道（UpdateTask），保证两端数据汇总到同一份本地缓存中处理。

com.alibaba.nacos.client.naming.remote.gprc.NamingPushRequestHandler

// 客户端接收服务实例推送的入口
public Response requestReply(Request request, Connection connection) {
NotifySubscriberRequest notifyRequest = (NotifySubscriberRequest) request;
// 委托ServiceInfoHolder处理推送的serviceInfo
serviceInfoHolder.processServiceInfo(notifyRequest.getServiceInfo());
return new NotifySubscriberResponse();
}

——————————————————————————————————————————————————————————————————————————————

// 处理服务端推送/查询返回的 ServiceInfo，被 NamingPushRequestHandler（推送通道）和 UpdateTask（定时拉取）共同调用。核心逻辑是对比新旧数据 ，有变更时写入缓存，然后发布 InstancesChangeEvent。
public ServiceInfo processServiceInfo(ServiceInfo serviceInfo) {

String serviceKey = serviceInfo.getKey();
if (serviceKey == null) { return null; }

// 从本地缓存中取出旧的 ServiceInfo
ServiceInfo oldService = serviceInfoMap.get(serviceInfo.getKey());

// 空推送保护：服务端推送了空实例列表且配置了保护，忽略本次推送
if (isEmptyOrErrorPush(serviceInfo)) {
return oldService; // 保留旧数据，跳过更新
}

// 更新本地缓存
serviceInfoMap.put(serviceInfo.getKey(), serviceInfo);

// 对比新旧数据，计算实例差异（added / removed / modified）
InstancesDiff diff = getServiceInfoDiff(oldService, serviceInfo);

// 有变更时，发布实例变更事件并写磁盘快照
if (diff.hasDifferent()) {
if (!failoverReactor.isFailoverSwitch(serviceKey)) {
NotifyCenter.publishEvent(
new InstancesChangeEvent(notifierEventScope,
serviceInfo.getName(), serviceInfo.getGroupName(),
serviceInfo.getClusters(), serviceInfo.getHosts(), diff));
}
DiskCache.write(serviceInfo, cacheDir); // 保存到本地磁盘快照
}

return serviceInfo;
}

——————————————————————————————————————————————————————————————————————————————

// 空推送保护,hosts=null或启用了空保护且数据校验不通过
private boolean isEmptyOrErrorPush(ServiceInfo serviceInfo) {
return null == serviceInfo.getHosts()
|| (pushEmptyProtection && !serviceInfo.validate());
}

5.3 事件监听回调：EventListener

InstancesChangeNotifier收到事件后按 groupName@@serviceName 匹配所有已注册的 NamingSelectorWrapper，逐个调用 notifyListener()。通过 NamingListenerInvoker.invoke() 调用用户 EventListener.onEvent()。若用户监听器是 AbstractEventListener 且提供了自定义线程池，则在用户线程池中异步执行，否则在 Nacos 推送线程中同步执行。

在nacos-example示例代码中，只有Dubbo 实现了EventListener，而OpenFeign却没有。根源是两者服务发现与本地缓存模型不同：Dubbo 消费端通过 RegistryDirectory 维护长生命周期 Invoker 地址池，Invoker 绑定 TCP 长连接资源、容器启动后常驻内存，每次 RPC 调用直接从内存列表负载均衡选节点，Nacos 客户端缓存更新无法同步 Dubbo 自有地址池，必须自定义 EventListener 接收 Nacos 实例变更推送、实时刷新 Invoker 列表，防止请求打到已下线实例；OpenFeign 搭配 Spring Cloud LoadBalancer，虽 LoadBalancer 具备短时 TTL 缓存，但不存在 Dubbo 式常驻连接地址池，每次 Feign 发起 HTTP 调用都会通过LoadBalancerClient重新查询实例，实例数据依托 Nacos 客户端内置ServiceInfoHolder，由 Nacos2.x gRPC 主动推送 + UpdateTask 定时兜底双机制自动刷新缓存，框架内部 NacosWatch 已隐式注册监听器，业务不用手动实现 EventListener；即便 LoadBalancer 短时缓存存在数据滞后，也可依靠断路器、重试机制做调用容错。

总结来说就是Dubbo需要外部事件回调主动同步常驻地址池；OpenFeign无长生命周期连接缓存，依赖Nacos客户端原生的事件 + 定时机制保障实例最终一致性。

com.alibaba.nacos.client.naming.event.InstancesChangeNotifier

// 收到实例变更事件后，按服务名匹配所有注册的NamingSelectorWrapper，逐个通知用户监听器
public void onEvent(InstancesChangeEvent event) {
String subId = NamingUtils.getGroupedName(event.getServiceName(), event.getGroupName());
Collection<NamingSelectorWrapper> selectorWrappers = selectorManager.getSelectorWrappers(subId);
for (NamingSelectorWrapper selectorWrapper : selectorWrappers) {
selectorWrapper.notifyListener(event);
}
}

——————————————————————————————————————————————————————————————————————————————

com.alibaba.nacos.client.naming.selector.NamingListenerInvoker

// 最终触发用户监听器回调。
public void invoke(NamingEvent event) {
// 如果用户监听器是AbstractEventListener且提供了自定义线程池，
// 则在用户线程池中异步执行回调，避免阻塞Nacos的推送线程
if (listener instanceof AbstractEventListener
&& ((AbstractEventListener) listener).getExecutor() != null) {
// 异步：将回调提交到用户指定的线程池
((AbstractEventListener) listener).getExecutor()
.execute(() -> listener.onEvent(event));
} else {
// 在Nacos的推送线程中直接执行回调
listener.onEvent(event);
}
}

六、全文小结

本文完整梳理了 Nacos 2.x 服务消费端订阅与实例发现的全链路流程。在Spring Cloud生态下，OpenFeign远程调用会经过 LoadBalancer负载均衡，最终由NacosDiscoveryClient作为统一入口完成服务实例查询。其核心查询方法selectInstances内置三级容灾降级策略，优先级依次为：本地 Failover 容灾文件兜底、订阅缓存读取、服务端直连查询，默认采用订阅缓存模式，兼顾高可用与查询性能。

服务订阅采用懒加载触发机制，首次调用服务时才执行 subscribe 订阅逻辑。一方面通过gRPC一元请求向服务端注册订阅关系、拉取全量实例数据并建立长连接推送通道；另一方面启动 UpdateTask 定时兜底任务，默认6秒主动轮询、失败指数退避，与服务端主动推送形成“推送为主、轮询兜底”的双保障机制，持续更新客户端 ServiceInfoHolder本地缓存。

服务端负责维护订阅关系与服务实例索引，通过ServiceStorage聚合实例数据。当服务实例发生上下线变更时，服务端基于事件驱动机制，通过延迟任务引擎合并推送任务，最终通过 gRPC 双向流向所有订阅客户端定向推送变更数据。客户端接收推送后，对比新旧实例数据、更新本地缓存与磁盘快照，数据变更时发布 InstancesChangeEvent 事件，触发监听器回调。

从框架差异来看，Dubbo与OpenFeign的监听实现逻辑截然不同：Dubbo 消费端维护长生命周期 Invoker 地址池，必须手动实现 EventListener 同步更新本地路由，避免调用下线实例；而 OpenFeign 依托 Spring Cloud LoadBalancer 每次调用重新获取实例，由 Nacos 框架自动维护缓存更新，无需业务手动实现监听器，短时缓存滞后可通过熔断、重试机制容错。整体整套机制实现了 Nacos 服务发现实时更新、故障容灾、最终一致的高可用能力。