当前位置：首页 > news >正文

Nacos源码—6.Nacos升级gRPC分析二

news 来源：原创 2025/8/11 15:40:24

大纲

1.Nacos 2.x版本的一些变化

2.客户端升级gRPC发起服务注册

3.服务端进行服务注册时的处理

4.客户端服务发现和服务端处理服务订阅的源码分析

(1)Nacos客户端进行服务发现的源码

(2)Nacos服务端处理服务订阅请求的源码

(1)Nacos客户端进行服务发现的源码

一.nacos-discovery引入Ribbon实现服务调用时的负载均衡

二.nacos-discovery如何整合Ribbon实现服务调用时的负载均衡

三.nacos-client如何进行服务发现

一.nacos-discovery引入Ribbon实现服务调用时的负载均衡

Nacos客户端就是引入了nacos-discovery + nacos-client依赖的项目。由于nacos-discovery整合了Ribbon，所以Ribbon可以调用Nacos服务端的服务实例查询列表接口。于是Nacos客户端便借助Ribbon实现了服务调用时的负载均衡，即Ribbon会从服务实例列表中选择一个服务实例给客户端进行服务调用。

在nacos-discovery的pom.xml中，可以看到它引入了Ribbon依赖：

二.nacos-discovery如何整合Ribbon实现服务调用时的负载均衡

在Ribbon中会有一个ServerList接口，如下所示：ServerList就是一个扩展接口，这个接口的作用就是获取Server列表。然后nacos-discovery会针对这个接口进行实现，从而整合Ribbon。

从引入的包来看，loadbalancer是属于Ribbon源码包下的。而LoadBalancer则是Ribbon中的负载均衡器。负载均衡器会结合IRule负载均衡策略，从服务实例列表中选择一个实例。

package com.netflix.loadbalancer;import java.util.List;//Interface that defines the methods sed to obtain the List of Servers
public interface ServerList<T extends Server> {public List<T> getInitialListOfServers();//Return updated list of servers. This is called say every 30 secspublic List<T> getUpdatedListOfServers();
}

当Nacos客户端进行微服务调用时，会通过Ribbon来选出一个服务实例，此时Ribbon会调用NacosServerList的getUpdatedListOfServers()方法获取服务实例列表。

nacos-discovery的NacosServerList类继承了AbstractServerList类，而且实现了Ribbon的ServerList接口的两个方法，如下所示：

public abstract class AbstractServerList<T extends Server> implements ServerList<T>, IClientConfigAware {......
}public class NacosServerList extends AbstractServerList<NacosServer> {private NacosDiscoveryProperties discoveryProperties;private String serviceId;public NacosServerList(NacosDiscoveryProperties discoveryProperties) {this.discoveryProperties = discoveryProperties;}@Overridepublic List<NacosServer> getInitialListOfServers() {return getServers();}@Overridepublic List<NacosServer> getUpdatedListOfServers() {return getServers();}private List<NacosServer> getServers() {try {//读取分组String group = discoveryProperties.getGroup();//通过服务名称、分组、true(表示只需要健康实例)，//调用NacosNamingService.selectInstances()方法来查询服务实例列表List<Instance> instances = discoveryProperties.namingServiceInstance().selectInstances(serviceId, group, true);//把Instance转换成NacosServer类型return instancesToServerList(instances);} catch (Exception e) {throw new IllegalStateException("Can not get service instances from nacos, serviceId=" + serviceId, e);}}private List<NacosServer> instancesToServerList(List<Instance> instances) {List<NacosServer> result = new ArrayList<>();if (CollectionUtils.isEmpty(instances)) {return result;}for (Instance instance : instances) {result.add(new NacosServer(instance));}return result;}public String getServiceId() {return serviceId;}@Overridepublic void initWithNiwsConfig(IClientConfig iClientConfig) {this.serviceId = iClientConfig.getClientName();}
}

NacosServerList的核心方法是getServers()，因为nacos-discovery实现Ribbon的两个接口都调用到了该方法。

在nacos-discovery的NacosServerList的getServers()方法中，会调用nacos-client的NacosNamingService的selectInstances()方法，来获取服务实例列表。

三.nacos-client如何进行服务发现

在nacos-client的NacosNamingService的selectInstances()方法中，首先会调用ServiceInfoHolder的getServiceInfo()方法从本地缓存获取数据。ServiceInfoHolder的serviceInfoMap中的value是一个ServiceInfo对象，在ServiceInfo对象中会有一个Listhosts属性来存放实例数据。

如果ServiceInfoHolder中的本地缓存没有对应的ServiceInfo对象，那么就会调用NamingClientProxyDelegate的subscribe()方法。该方法首先会开启一个查询服务实例列表的延时执行的任务，然后通过Client对象发送订阅请求，去服务端实时获取服务实例数据。

具体来说就是先调用ServiceInfoUpdateService的scheduleUpdateIfAbsent()方法，开启一个延迟执行查询服务实例列表的UpdateTask任务，然后再次调用ServiceInfoHolder的getServiceInfoMap()方法查询本地缓存。如果本地缓存为空，则向服务端发起gRPC请求获取服务实例数据，也就是通过调用NamingGrpcClientProxy的subscribe()方法，触发调用NamingGrpcClientProxy的doSubscribe()方法，再触发调用NamingGrpcClientProxy的requestToServer()方法，接着调用RpcClient的request()方法发送gRPC请求给服务端，最后调用ServiceInfoHolder的processServiceInfo()方法更新本地缓存。

public class NacosNamingService implements NamingService {private ServiceInfoHolder serviceInfoHolder;private NamingClientProxy clientProxy;...private void init(Properties properties) throws NacosException {...this.serviceInfoHolder = new ServiceInfoHolder(namespace, properties);this.clientProxy = new NamingClientProxyDelegate(this.namespace, serviceInfoHolder, properties, changeNotifier);}@Overridepublic List<Instance> selectInstances(String serviceName, String groupName, List<String> clusters, boolean healthy, boolean subscribe) throws NacosException {ServiceInfo serviceInfo;String clusterString = StringUtils.join(clusters, ",");//判断是否需要订阅，默认为trueif (subscribe) {//查询Nacos本地缓存数据serviceInfo = serviceInfoHolder.getServiceInfo(serviceName, groupName, clusterString);if (null == serviceInfo) {//如果本地缓存数据为空，则调用Client代理NamingClientProxyDelegate的订阅方法subscribe()，通过Client对象请求服务端获取数据serviceInfo = clientProxy.subscribe(serviceName, groupName, clusterString);}} else {serviceInfo = clientProxy.queryInstancesOfService(serviceName, groupName, clusterString, 0, false);}//返回数据return selectInstances(serviceInfo, healthy);}...
}//Naming client service information holder.
public class ServiceInfoHolder implements Closeable {private final ConcurrentMap<String, ServiceInfo> serviceInfoMap;...public ServiceInfo getServiceInfo(final String serviceName, final String groupName, final String clusters) {NAMING_LOGGER.debug("failover-mode: {}", failoverReactor.isFailoverSwitch());//获取keyString groupedServiceName = NamingUtils.getGroupedName(serviceName, groupName);String key = ServiceInfo.getKey(groupedServiceName, clusters);if (failoverReactor.isFailoverSwitch()) {return failoverReactor.getService(key);}//通过key从本地缓存中获取数据return serviceInfoMap.get(key);}...
}public class NamingClientProxyDelegate implements NamingClientProxy {private final ServiceInfoUpdateService serviceInfoUpdateService;private final ServiceInfoHolder serviceInfoHolder;private final NamingGrpcClientProxy grpcClientProxy;...public NamingClientProxyDelegate(String namespace, ServiceInfoHolder serviceInfoHolder, Properties properties, InstancesChangeNotifier changeNotifier) throws NacosException {...this.serviceInfoUpdateService = new ServiceInfoUpdateService(properties, serviceInfoHolder, this, changeNotifier);this.serviceInfoHolder = serviceInfoHolder;this.grpcClientProxy = new NamingGrpcClientProxy(namespace, securityProxy, serverListManager, properties, serviceInfoHolder);}...@Overridepublic ServiceInfo subscribe(String serviceName, String groupName, String clusters) throws NacosException {NAMING_LOGGER.info("[SUBSCRIBE-SERVICE] service:{}, group:{}, clusters:{} ", serviceName, groupName, clusters);String serviceNameWithGroup = NamingUtils.getGroupedName(serviceName, groupName);String serviceKey = ServiceInfo.getKey(serviceNameWithGroup, clusters);//1.调用ServiceInfoUpdateService.scheduleUpdateIfAbsent()方法开启一个查询服务实例列表的定时任务serviceInfoUpdateService.scheduleUpdateIfAbsent(serviceName, groupName, clusters);//再次查询本地的缓存数据ServiceInfo result = serviceInfoHolder.getServiceInfoMap().get(serviceKey);if (null == result || !isSubscribed(serviceName, groupName, clusters)) {//2.如果本地缓存还是为空，则使用gRPC来请求服务端，也就是调用NamingGrpcClientProxy.subscribe()方法result = grpcClientProxy.subscribe(serviceName, groupName, clusters);}//更新本地缓存serviceInfoHolder.processServiceInfo(result);return result;}...
}public class NamingGrpcClientProxy extends AbstractNamingClientProxy {...private final RpcClient rpcClient;private final NamingGrpcRedoService redoService;public NamingGrpcClientProxy(String namespaceId, SecurityProxy securityProxy, ServerListFactory serverListFactory, Properties properties, ServiceInfoHolder serviceInfoHolder) throws NacosException {...//通过RpcClientFactory.createClient()方法创建一个GrpcSdkClient对象实例，然后赋值给rpcClient属性this.rpcClient = RpcClientFactory.createClient(uuid, ConnectionType.GRPC, labels);this.redoService = new NamingGrpcRedoService(this);start(serverListFactory, serviceInfoHolder);}private void start(ServerListFactory serverListFactory, ServiceInfoHolder serviceInfoHolder) throws NacosException {rpcClient.serverListFactory(serverListFactory);rpcClient.registerConnectionListener(redoService);rpcClient.registerServerRequestHandler(new NamingPushRequestHandler(serviceInfoHolder));rpcClient.start();NotifyCenter.registerSubscriber(this);}...@Overridepublic ServiceInfo subscribe(String serviceName, String groupName, String clusters) throws NacosException {if (NAMING_LOGGER.isDebugEnabled()) {NAMING_LOGGER.debug("[GRPC-SUBSCRIBE] service:{}, group:{}, cluster:{} ", serviceName, groupName, clusters);}redoService.cacheSubscriberForRedo(serviceName, groupName, clusters);//执行订阅return doSubscribe(serviceName, groupName, clusters);}//Execute subscribe operation.public ServiceInfo doSubscribe(String serviceName, String groupName, String clusters) throws NacosException {//构建订阅服务请求——SubscribeServiceRequest对象SubscribeServiceRequest request = new SubscribeServiceRequest(namespaceId, groupName, serviceName, clusters, true);//向服务端发送SubscribeServiceRequest类型的请求SubscribeServiceResponse response = requestToServer(request, SubscribeServiceResponse.class);redoService.subscriberRegistered(serviceName, groupName, clusters);return response.getServiceInfo();}private <T extends Response> T requestToServer(AbstractNamingRequest request, Class<T> responseClass) throws NacosException {try {request.putAllHeader(getSecurityHeaders(request.getNamespace(), request.getGroupName(), request.getServiceName()));//实际会调用RpcClient.request()方法发起gRPC请求Response response = requestTimeout < 0 ? rpcClient.request(request) : rpcClient.request(request, requestTimeout);if (ResponseCode.SUCCESS.getCode() != response.getResultCode()) {throw new NacosException(response.getErrorCode(), response.getMessage());}if (responseClass.isAssignableFrom(response.getClass())) {return (T) response;}NAMING_LOGGER.error("Server return unexpected response '{}', expected response should be '{}'", response.getClass().getName(), responseClass.getName());} catch (Exception e) {throw new NacosException(NacosException.SERVER_ERROR, "Request nacos server failed: ", e);}throw new NacosException(NacosException.SERVER_ERROR, "Server return invalid response");}...
}

其中UpdateTask任务的run()方法会先调用NamingClientProxy.queryInstancesOfService()方法，然后调用ServiceInfoHolder的processServiceInfo()方法向服务端查询服务实例列表以及更新本地服务实例缓存。当该任务执行完毕时，会继续向调度线程池提交一个延迟6s执行的任务，从而实现不断更新本地缓存的服务实例列表。

在ServiceInfoHolder的processServiceInfo()方法更新本地服务实例缓存中，会判断服务实例是否发生改变。如果有改变，那么客户端会先发布一个服务实例改变事件InstancesChangeEvent，然后把新的服务实例数据写入本地磁盘。

public class ServiceInfoUpdateService implements Closeable {private final NamingClientProxy namingClientProxy;private final Map<String, ScheduledFuture<?>> futureMap = new HashMap<String, ScheduledFuture<?>>();...//Schedule update if absent.public void scheduleUpdateIfAbsent(String serviceName, String groupName, String clusters) {//生成一个serverKeyString serviceKey = ServiceInfo.getKey(NamingUtils.getGroupedName(serviceName, groupName), clusters);//判断当前serviceKey是否已经开启了对应的定时任务，如果已经开启就不开启了if (futureMap.get(serviceKey) != null) {return;}//加一把同步锁，避免并发冲突synchronized (futureMap) {//加锁之后再进行一次判断，双重检测Double Checkif (futureMap.get(serviceKey) != null) {return;}//添加UpdateTask任务到executor延迟执行ScheduledFuture<?> future = addTask(new UpdateTask(serviceName, groupName, clusters));futureMap.put(serviceKey, future);}}private synchronized ScheduledFuture<?> addTask(UpdateTask task) {return executor.schedule(task, DEFAULT_DELAY, TimeUnit.MILLISECONDS);}...public class UpdateTask implements Runnable {long lastRefTime = Long.MAX_VALUE;private boolean isCancel;private final String serviceName;private final String groupName;private final String clusters;private final String groupedServiceName;private final String serviceKey;//the fail situation. 1:can't connect to server 2:serviceInfo's hosts is emptyprivate int failCount = 0;public UpdateTask(String serviceName, String groupName, String clusters) {this.serviceName = serviceName;this.groupName = groupName;this.clusters = clusters;this.groupedServiceName = NamingUtils.getGroupedName(serviceName, groupName);this.serviceKey = ServiceInfo.getKey(groupedServiceName, clusters);}@Overridepublic void run() {long delayTime = DEFAULT_DELAY;try {if (!changeNotifier.isSubscribed(groupName, serviceName, clusters) && !futureMap.containsKey(serviceKey)) {NAMING_LOGGER.info("update task is stopped, service:{}, clusters:{}", groupedServiceName, clusters);isCancel = true;return;}//获取本地缓存ServiceInfo serviceObj = serviceInfoHolder.getServiceInfoMap().get(serviceKey);if (serviceObj == null) {//如果本地缓存为空，则通过gRPC去查询服务端的服务实例数据serviceObj = namingClientProxy.queryInstancesOfService(serviceName, groupName, clusters, 0, false);//更新本地缓存serviceInfoHolder.processServiceInfo(serviceObj);//更新UpdateTask的lastRefTime属性，即更新UpdateTask任务的数据获取时间lastRefTime = serviceObj.getLastRefTime();return;}//如果本地缓存不为空，则判断本地缓存最后一次刷新时间 是否小于等于 最后一次UpdateTask任务的数据获取时间if (serviceObj.getLastRefTime() <= lastRefTime) {//如果小于等于，则重新查询服务端的服务实例数据serviceObj = namingClientProxy.queryInstancesOfService(serviceName, groupName, clusters, 0, false);//更新本地缓存serviceInfoHolder.processServiceInfo(serviceObj);}lastRefTime = serviceObj.getLastRefTime();if (CollectionUtils.isEmpty(serviceObj.getHosts())) {incFailCount();return;}//TODO multiple time can be configured.//计算下一次定时任务执行的时间，这个时间默认是6sdelayTime = serviceObj.getCacheMillis() * DEFAULT_UPDATE_CACHE_TIME_MULTIPLE;//请求成功后，重置错误次数为0resetFailCount();} catch (Throwable e) {//记录请求失败的次数incFailCount();NAMING_LOGGER.warn("[NA] failed to update serviceName: {}", groupedServiceName, e);} finally {if (!isCancel) {//根据请求失败的次数，动态调整重新提交的UpdateTask定时任务的执行时间executor.schedule(this, Math.min(delayTime << failCount, DEFAULT_DELAY * 60), TimeUnit.MILLISECONDS);}}}private void incFailCount() {int limit = 6;if (failCount == limit) {return;}failCount++;}private void resetFailCount() {failCount = 0;}}...
}//Naming client service information holder.
public class ServiceInfoHolder implements Closeable {private final ConcurrentMap<String, ServiceInfo> serviceInfoMap;...//Process service info.public ServiceInfo processServiceInfo(ServiceInfo serviceInfo) {String serviceKey = serviceInfo.getKey();if (serviceKey == null) {return null;}//获取本地缓存中的服务实例ServiceInfo oldService = serviceInfoMap.get(serviceInfo.getKey());if (isEmptyOrErrorPush(serviceInfo)) {return oldService;}//更新本地缓存中的服务实例serviceInfoMap.put(serviceInfo.getKey(), serviceInfo);//判断服务实例是否有改变boolean changed = isChangedServiceInfo(oldService, serviceInfo);if (StringUtils.isBlank(serviceInfo.getJsonFromServer())) {serviceInfo.setJsonFromServer(JacksonUtils.toJson(serviceInfo));}MetricsMonitor.getServiceInfoMapSizeMonitor().set(serviceInfoMap.size());if (changed) {NAMING_LOGGER.info("current ips:({}) service: {} -> {}", serviceInfo.ipCount(), serviceInfo.getKey(), JacksonUtils.toJson(serviceInfo.getHosts()));//发布服务实例改变事件NotifyCenter.publishEvent(new InstancesChangeEvent(serviceInfo.getName(), serviceInfo.getGroupName(), serviceInfo.getClusters(), serviceInfo.getHosts()));//将服务实例信息写入本地磁盘DiskCache.write(serviceInfo, cacheDir);}return serviceInfo;}...
}

(2)Nacos服务端处理服务订阅请求的源码

一.根据要查询的Service对象读取缓存

二.添加订阅者即Subscriber对象

假设order-service需要调用stock-service的接口，那么order-service(Nacos客户端)就要向Nacos服务端订阅stock-service服务，也就是order-service需要从服务端获取到stock-service的所有服务实例。

客户端向服务端发起订阅请求的参数类型是SubscribeServiceRequest。服务订阅请求的处理方法是SubscribeServiceRequestHandler的handle()，该方法会先从SubscribeServiceRequest对象里获取信息构建Service对象，然后再根据RequestMeta请求元数据构建Subscriber对象，接着就会调用ServiceStorage的getData()方法读取缓存中的服务实例，以及通过clientOperationService的subscribeService()方法添加订阅者。

//Handler to handle subscribe service.
@Component
public class SubscribeServiceRequestHandler extends RequestHandler<SubscribeServiceRequest, SubscribeServiceResponse> {private final ServiceStorage serviceStorage;private final EphemeralClientOperationServiceImpl clientOperationService;...//假设order-service需要调用stock-service的接口，那么order-service(Nacos客户端)就要向服务端订阅stock-service服务//也就是order-service需要从服务端获取到(查询出)stock-service的所有服务实例@Override@Secured(action = ActionTypes.READ)public SubscribeServiceResponse handle(SubscribeServiceRequest request, RequestMeta meta) throws NacosException {String namespaceId = request.getNamespace();String serviceName = request.getServiceName();String groupName = request.getGroupName();String app = request.getHeader("app", "unknown");String groupedServiceName = NamingUtils.getGroupedName(serviceName, groupName);//构建要查询的Service服务对象，对应的是stock-serivceService service = Service.newService(namespaceId, groupName, serviceName, true);//构建要订阅Service服务的订阅者，对应的是order-serviceSubscriber subscriber = new Subscriber(meta.getClientIp(), meta.getClientVersion(), app, meta.getClientIp(), namespaceId, groupedServiceName, 0, request.getClusters());//1.调用ServiceStorage.getData()方法读取缓存ServiceInfo serviceInfo = ServiceUtil.selectInstancesWithHealthyProtection(serviceStorage.getData(service), metadataManager.getServiceMetadata(service).orElse(null), subscriber);if (request.isSubscribe()) {//2.添加订阅者，如果订阅的服务有变动，则需要通知订阅者clientOperationService.subscribeService(service, subscriber, meta.getConnectionId());} else {clientOperationService.unsubscribeService(service, subscriber, meta.getConnectionId());}return new SubscribeServiceResponse(ResponseCode.SUCCESS.getCode(), "success", serviceInfo);}}

一.根据要查询的Service对象读取缓存

调用的方法是ServiceStorage.getData()：

-> serviceDataIndexes.get()
-> ServiceStorage.getPushData()方法
-> ServiceStorage.emptyServiceInfo()方法
-> ServiceStorage.getAllInstancesFromIndex()方法
-> ClientServiceIndexesManager.getAllClientsRegisteredService()方法 
-> ServiceStorage.getInstanceInfo()方法根据clientId获取Instance对象
-> EphemeralIpPortClientManager.getClient()方法
-> AbstractClient.getInstancePublishInfo()方法
-> ServiceStorage.parseInstance()方法
-> serviceDataIndexes.put() + serviceClusterIndex.put()

ServiceStorage的getData()方法在读取缓存时，获取要查询的Service服务对象下的全部Instance实例会分三步：一是从注册表中获取要查询的Service对象下的全部clientId，二是根据clientId获取对应的Client对象，三是根据Client对象获取对应的Instance信息。

@Component
public class ServiceStorage {//缓存要查询的Service服务对象对应的已注册的服务详情private final ConcurrentMap<Service, ServiceInfo> serviceDataIndexes;//缓存要查询的Service服务对象对应的服务集群private final ConcurrentMap<Service, Set<String>> serviceClusterIndex;...public ServiceInfo getData(Service service) {//判断缓存中是否有数据，传入的参数service就是要查询的Service服务对象，对应的是stock-serivce//如果有则直接读取缓存数据，如果没有则调用ServiceStorage.getPushData(service)return serviceDataIndexes.containsKey(service) ? serviceDataIndexes.get(service) : getPushData(service);}public ServiceInfo getPushData(Service service) {//调用ServiceStorage.emptyServiceInfo()方法创建空的ServiceInfo对象ServiceInfo result = emptyServiceInfo(service);if (!ServiceManager.getInstance().containSingleton(service)) {return result;}//调用ServiceStorage.getAllInstancesFromIndex()方法获服务取实例列表//ServiceInfo的hosts属性就包含了该服务的所有Instance实例数据result.setHosts(getAllInstancesFromIndex(service));//将获取到的ServiceInfo对象放入到缓存中serviceDataIndexes.put(service, result);return result;}private ServiceInfo emptyServiceInfo(Service service) {ServiceInfo result = new ServiceInfo();result.setName(service.getName());result.setGroupName(service.getGroup());result.setLastRefTime(System.currentTimeMillis());result.setCacheMillis(switchDomain.getDefaultPushCacheMillis());return result;}private List<Instance> getAllInstancesFromIndex(Service service) {//传入的参数service就是要查询的Service服务对象，对应的是stock-serivceSet<Instance> result = new HashSet<>();Set<String> clusters = new HashSet<>();//each = clientId，调用ClientServiceIndexesManager.getAllClientsRegisteredService()方法获取Service服务对象的所有clientIdfor (String each : serviceIndexesManager.getAllClientsRegisteredService(service)) {//调用ServiceStorage.getInstanceInfo()方法，根据clientId查询出对应的Instance信息，即InstancePublishInfo对象Optional<InstancePublishInfo> instancePublishInfo = getInstanceInfo(each, service);if (instancePublishInfo.isPresent()) {//对象转换，将InstancePublishInfo对象转换成Instance对象Instance instance = parseInstance(service, instancePublishInfo.get());result.add(instance);clusters.add(instance.getClusterName());}}//cache clusters of this service//缓存要查询的Service服务对象，对应有哪些集群serviceClusterIndex.put(service, clusters);return new LinkedList<>(result);}private Optional<InstancePublishInfo> getInstanceInfo(String clientId, Service service) {//通过调用EphemeralIpPortClientManager.getClient()方法，根据clientId获取对应的Client连接对象Client client = clientManager.getClient(clientId);if (null == client) {return Optional.empty();}//调用AbstractClient.getInstancePublishInfo()方法，获取该Client客户端注册的服务实例信息InstancePublishInforeturn Optional.ofNullable(client.getInstancePublishInfo(service));}private Instance parseInstance(Service service, InstancePublishInfo instanceInfo) {Instance result = InstanceUtil.parseToApiInstance(service, instanceInfo);Optional<InstanceMetadata> metadata = metadataManager.getInstanceMetadata(service, instanceInfo.getMetadataId());metadata.ifPresent(instanceMetadata -> InstanceUtil.updateInstanceMetadata(result, instanceMetadata));return result;}...
}@Component
public class ClientServiceIndexesManager extends SmartSubscriber {//注册表(服务提供者)，一个Service服务对象，对应多个服务实例的clientIdprivate final ConcurrentMap<Service, Set<String>> publisherIndexes = new ConcurrentHashMap<>();...public Collection<String> getAllClientsRegisteredService(Service service) {//服务端在处理服务注册时，最后会调用ClientServiceIndexesManager.addPublisherIndexes()方法：//将客户端对应的clientId放入到publisherIndexes注册表中//所有下面的代码会从注册表中获取要查询的Service服务对象对应的clientId集合return publisherIndexes.containsKey(service) ? publisherIndexes.get(service) : new ConcurrentHashSet<>();}private void addPublisherIndexes(Service service, String clientId) {//判断注册表是否存在该Service，不存在则创建一个空的ConcurrentHashSetpublisherIndexes.computeIfAbsent(service, (key) -> new ConcurrentHashSet<>());//把clientId放入到对应的Service中publisherIndexes.get(service).add(clientId);//发布服务改变事件NotifyCenter.publishEvent(new ServiceEvent.ServiceChangedEvent(service, true));}...
}@Component("ephemeralIpPortClientManager")
public class EphemeralIpPortClientManager implements ClientManager {//key是请求参数中的connectionId即clientId，value是一个继承了实现Client接口的AbstractClient的IpPortBasedClient对象private final ConcurrentMap<String, IpPortBasedClient> clients = new ConcurrentHashMap<>();...@Overridepublic Client getClient(String clientId) {//客户端和服务端建立长连接时，就会通过EphemeralIpPortClientManager.clientConnected()方法将clientId放入到clientsreturn clients.get(clientId);}...
}public abstract class AbstractClient implements Client {//publishers其实就是记录该客户端提供的服务和服务实例，一个客户端可提供多个服务//存储客户端发送过来的请求中的Instance信息，当然这些信息已封装为InstancePublishInfo对象//key为已注册的Service，value是根据请求中的instance实例信息封装的InstancePublishInfo对象protected final ConcurrentHashMap<Service, InstancePublishInfo> publishers = new ConcurrentHashMap<>(16, 0.75f, 1);...@Overridepublic InstancePublishInfo getInstancePublishInfo(Service service) {return publishers.get(service);}...
}

二.添加订阅者即Subscriber对象

调用的方法是EphemeralClientOperationServiceImpl.subscribeService()，添加订阅者其实就是先根据clientId找出对应的客户端Client对象，然后往AbstractClient.subscribers属性放入服务对象和对应的订阅者对象，最后再发布一个客户端订阅服务事件ClientSubscribeServiceEvent。

这个事件会被ClientServiceIndexesManager的onEvent()方法处理，即调用ClientServiceIndexesManager的addSubscriberIndexes()方法，该方法会继续发布一个服务订阅事件ServiceSubscribedEvent。

//Operation service for ephemeral clients and services.
@Component("ephemeralClientOperationService")
public class EphemeralClientOperationServiceImpl implements ClientOperationService {...//添加订阅者//@param service    service：要查询的Service对象，比如stock-service//@param subscriber subscribe：订阅者，比如对应order-service//@param clientId   id of client：对应order-service与Nacos服务端的连接ID@Overridepublic void subscribeService(Service service, Subscriber subscriber, String clientId) {//传入的service是要查询的Service对象，比如stock-serviceService singleton = ServiceManager.getInstance().getSingletonIfExist(service).orElse(service);//传入的clientId是代表着order-service的Client对象，调用EphemeralIpPortClientManager.getClient()方法Client client = clientManager.getClient(clientId);if (!clientIsLegal(client, clientId)) {return;}//往代表着order-service的Client对象中，添加订阅者client.addServiceSubscriber(singleton, subscriber);client.setLastUpdatedTime();//发布客户端订阅服务事件ClientSubscribeServiceEvent，也就是order-service客户端订阅了service服务NotifyCenter.publishEvent(new ClientOperationEvent.ClientSubscribeServiceEvent(singleton, clientId));}...
}@Component("ephemeralIpPortClientManager")
public class EphemeralIpPortClientManager implements ClientManager {//key是请求参数中的connectionId即clientId，value是一个继承了实现Client接口的AbstractClient的IpPortBasedClient对象private final ConcurrentMap<String, IpPortBasedClient> clients = new ConcurrentHashMap<>();...@Overridepublic Client getClient(String clientId) {//客户端和服务端建立长连接时，就会通过EphemeralIpPortClientManager.clientConnected()方法将clientId放入到clientsreturn clients.get(clientId);}...
}public abstract class AbstractClient implements Client {//subscribers存放着：订阅者Subscriber(其实可理解为当前客户端)订阅了的Service服务//subscribers的key=stock-service(要订阅的某个服务)、value=order-service(订阅者，某个具体的包含IP的服务实例)protected final ConcurrentHashMap<Service, Subscriber> subscribers = new ConcurrentHashMap<>(16, 0.75f, 1);...@Overridepublic boolean addServiceSubscriber(Service service, Subscriber subscriber) {//服务订阅时，添加订阅者//subscribers的key=stock-service(要订阅的某个服务)、value=order-service(订阅者，某个具体的包含IP的服务实例)if (null == subscribers.put(service, subscriber)) {MetricsMonitor.incrementSubscribeCount();}return true;}...
}@Component
public class ClientServiceIndexesManager extends SmartSubscriber {//注册表(服务提供者)，一个Service服务对象，对应多个服务实例的clientIdprivate final ConcurrentMap<Service, Set<String>> publisherIndexes = new ConcurrentHashMap<>();//订阅者列表(服务消费者)，一个Service服务对象，对应多个订阅者的clientIdprivate final ConcurrentMap<Service, Set<String>> subscriberIndexes = new ConcurrentHashMap<>();...//可以处理客户端注册事件ClientRegisterServiceEvent@Overridepublic void onEvent(Event event) {if (event instanceof ClientEvent.ClientDisconnectEvent) {handleClientDisconnect((ClientEvent.ClientDisconnectEvent) event);} else if (event instanceof ClientOperationEvent) {handleClientOperation((ClientOperationEvent) event);}}private void handleClientOperation(ClientOperationEvent event) {Service service = event.getService();String clientId = event.getClientId();if (event instanceof ClientOperationEvent.ClientRegisterServiceEvent) {//处理客户端注册事件ClientRegisterServiceEventaddPublisherIndexes(service, clientId);} else if (event instanceof ClientOperationEvent.ClientDeregisterServiceEvent) {//处理客户端注销事件ClientDeregisterServiceEventremovePublisherIndexes(service, clientId);} else if (event instanceof ClientOperationEvent.ClientSubscribeServiceEvent) {//处理客户端订阅服务事件ClientSubscribeServiceEventaddSubscriberIndexes(service, clientId);} else if (event instanceof ClientOperationEvent.ClientUnsubscribeServiceEvent) {//处理客户端取消订阅事件ClientUnsubscribeServiceEventremoveSubscriberIndexes(service, clientId);}}private void addSubscriberIndexes(Service service, String clientId) {//传入的service是要查询的Service对象stock-service，clientId是订阅者order-service对应的客户端连接对象IDsubscriberIndexes.computeIfAbsent(service, (key) -> new ConcurrentHashSet<>());//Fix #5404, Only first time add need notify event. 只有第一次添加时需要发布通知事件if (subscriberIndexes.get(service).add(clientId)) {//发布服务订阅事件ServiceSubscribedEventNotifyCenter.publishEvent(new ServiceEvent.ServiceSubscribedEvent(service, clientId));}}...
}

(3)总结

相关文章：