Nacos源码—6.Nacos升级gRPC分析一
技术分享
大纲
1.Nacos 2.x版本的一些变化
2.客户端升级gRPC发起服务注册
3.服务端进行服务注册时的处理
4.客户端服务发现和服务端处理服务订阅的源码分析
1.Nacos 2.x版本的一些变化
变化一：客户端和服务端的交互方式由HTTP升级为gRPC
Nacos 1.x服务端会提供一系列的HTTP接口供客户端请求调用，Nacos 2.x服务端会定义一些列Handler处理类来处理客户端的gRPC请求。
Nacos 1.x进行服务注册时使用的是HTTP短连接，Nacos 2.x进行服务注册时使用的是RPC长连接。
变化二：注册中心的注册表由双重Map结构变为轻量的一个Map
Nacos 1.x版本中的注册中心是使用双重Map结构来存储注册表的，这样使得注册表还是比较重量级的，在并发高时也需要考虑并发冲突。
Nacos 2.x版本则把注册表轻量化了，服务端在处理服务注册时，只是简单地往一个Map记写入客户端连接的ID。
变化三：大量使用了事件驱动
Nacos 2.x版本的Nacos里，使用了非常多的事件驱动。比如服务注册、服务销毁、服务变更等都先通过通知中心来发布一个事件，然后通过处理事件来来处理后续的逻辑流程。
2.客户端升级gRPC发起服务注册
(1)客户端和服务端的版本选择
(2)Nacos客户端项目启动时自动触发服务实例注册
(3)Nacos客户端通过gRPC方式发起服务实例注册
(4)总结
(1)客户端和服务端的版本选择
<!-- 使用的是Nacos 1.4.1 --> <!-- <properties> <java.version>1.8</java.version> <spring-cloud.version>Hoxton.SR8</spring-cloud.version> <spring-cloud-alibaba.version>2.2.5.RELEASE</spring-cloud-alibaba.version> </properties> --> <!-- 使用的是Nacos 2.1.0 --> <properties> <java.version>1.8</java.version> <spring-cloud.version>Hoxton.SR12</spring-cloud.version> <spring-cloud-alibaba.version>2.2.8.RELEASE</spring-cloud-alibaba.version> </properties>(2)Nacos客户端项目启动时自动触发服务实例注册
spring-cloud-starter-alibaba-nacos-discovery依赖包会自动注册服务。查看这个依赖包中的spring.factories文件，会指定一些Configuration类。Spring Boot启动时会扫描spring.factories文件，然后创建里面的配置类。
在spring.pactories文件中，与注册相关的类就是：NacosServiceRegistryAutoConfiguration这个Nacos服务注册自动配置类。
Nacos服务注册自动配置类NacosServiceRegistryAutoConfiguration如下，该配置类创建了三个Bean：
第一个Bean：NacosServiceRegistry
这个Bean在创建时，会传入加载了yml配置文件内容的类NacosDiscoveryProperties。
第二个Bean：NacosRegistration
这个Bean在创建时，会传入加载了yml配置文件内容的类NacosDiscoveryProperties。
第三个Bean：NacosAutoServiceRegistration
这个Bean在创建时，会传入NacosServiceRegistry和NacosRegistration两个Bean，然后该Bean继承了AbstractAutoServiceRegistration抽象类。该抽象类实现了ApplicationListener接口，所以项目启动时便是利用了Spring的监听事件来实现自动注册服务的。因为在Spring容器启动的最后会执行finishRefresh()方法，然后会发布一个事件，该事件会触发调用onApplicationEvent()方法。
调用AbstractAutoServiceRegistration的onApplicationEvent()方法时，首先会调用AbstractAutoServiceRegistration的bind()方法，然后调用AbstractAutoServiceRegistration的start()方法，接着调用AbstractAutoServiceRegistration的register()方法发起注册，也就是调用this.serviceRegistry的register()方法完成服务注册的具体工作。
而AbstractAutoServiceRegistration的serviceRegistry属性，是在服务注册自动配置类NacosServiceRegistryAutoConfiguration，创建第三个Bean—NacosAutoServiceRegistration时，通过传入其创建的第一个Bean—NacosServiceRegistry进行赋值的。
@Configuration(proxyBeanMethods = false) @EnableConfigurationProperties @ConditionalOnNacosDiscoveryEnabled @ConditionalOnProperty(value = "spring.cloud.service-registry.auto-registration.enabled", matchIfMissing = true) @AutoConfigureAfter({ AutoServiceRegistrationConfiguration.class, AutoServiceRegistrationAutoConfiguration.class, NacosDiscoveryAutoConfiguration.class }) public class NacosServiceRegistryAutoConfiguration { @Bean public NacosServiceRegistry nacosServiceRegistry(NacosServiceManager nacosServiceManager, NacosDiscoveryProperties nacosDiscoveryProperties) { return new NacosServiceRegistry(nacosServiceManager, nacosDiscoveryProperties); } @Bean @ConditionalOnBean(AutoServiceRegistrationProperties.class) public NacosRegistration nacosRegistration(ObjectProvider<List<NacosRegistrationCustomizer>> registrationCustomizers, NacosDiscoveryProperties nacosDiscoveryProperties, ApplicationContext context) { return new NacosRegistration(registrationCustomizers.getIfAvailable(), nacosDiscoveryProperties, context); } @Bean @ConditionalOnBean(AutoServiceRegistrationProperties.class) public NacosAutoServiceRegistration nacosAutoServiceRegistration(NacosServiceRegistry registry, AutoServiceRegistrationProperties autoServiceRegistrationProperties, NacosRegistration registration) { return new NacosAutoServiceRegistration(registry, autoServiceRegistrationProperties, registration); } } public class NacosAutoServiceRegistration extends AbstractAutoServiceRegistration<Registration> { ... private NacosRegistration registration; public NacosAutoServiceRegistration(ServiceRegistry<Registration> serviceRegistry, AutoServiceRegistrationProperties autoServiceRegistrationProperties, NacosRegistration registration) { super(serviceRegistry, autoServiceRegistrationProperties); this.registration = registration; } ... } public abstract class AbstractAutoServiceRegistration<R extends Registration> implements AutoServiceRegistration, ApplicationContextAware, ApplicationListener<WebServerInitializedEvent> { ... private final ServiceRegistry<R> serviceRegistry; private AutoServiceRegistrationProperties properties; protected AbstractAutoServiceRegistration(ServiceRegistry<R> serviceRegistry, AutoServiceRegistrationProperties properties) { this.serviceRegistry = serviceRegistry; this.properties = properties; } ... @Override @SuppressWarnings("deprecation") public void onApplicationEvent(WebServerInitializedEvent event) { bind(event); } public void bind(WebServerInitializedEvent event) { ApplicationContext context = event.getApplicationContext(); if (context instanceof ConfigurableWebServerApplicationContext) { if ("management".equals(((ConfigurableWebServerApplicationContext) context).getServerNamespace())) { return; } } this.port.compareAndSet(0, event.getWebServer().getPort()); this.start(); } public void start() { if (!isEnabled()) { if (logger.isDebugEnabled()) { logger.debug("Discovery Lifecycle disabled. Not starting"); } return; } if (!this.running.get()) { this.context.publishEvent(new InstancePreRegisteredEvent(this, getRegistration())); //发起注册 register(); if (shouldRegisterManagement()) { registerManagement(); } this.context.publishEvent(new InstanceRegisteredEvent<>(this, getConfiguration())); this.running.compareAndSet(false, true); } } protected void register() { //调用创建NacosAutoServiceRegistration时传入的NacosServiceRegistry实例的register()方法 this.serviceRegistry.register(getRegistration()); } ... } public class NacosServiceRegistry implements ServiceRegistry<Registration> { private final NacosDiscoveryProperties nacosDiscoveryProperties; public NacosServiceRegistry(NacosDiscoveryProperties nacosDiscoveryProperties) { this.nacosDiscoveryProperties = nacosDiscoveryProperties; } @Override public void register(Registration registration) { if (StringUtils.isEmpty(registration.getServiceId())) { log.warn("No service to register for nacos client..."); return; } NamingService namingService = namingService(); //获取服务ID、分组 String serviceId = registration.getServiceId(); String group = nacosDiscoveryProperties.getGroup(); //创建Instance对象 Instance instance = getNacosInstanceFromRegistration(registration); try { //发起服务实例注册 namingService.registerInstance(serviceId, group, instance); log.info("nacos registry, {} {} {}:{} register finished", group, serviceId, instance.getIp(), instance.getPort()); } catch (Exception e) { if (nacosDiscoveryProperties.isFailFast()) { log.error("nacos registry, {} register failed...{},", serviceId, registration.toString(), e); rethrowRuntimeException(e); } else { log.warn("Failfast is false. {} register failed...{},", serviceId, registration.toString(), e); } } } ... }总结：
Nacos客户端项目启动时自动触发注册服务实例的流程：Spring监听器调用onApplicationEvent() -> bind() -> start() -> register()，最后register()方法会调用serviceRegistry属性的register()方法进行注册。
整个流程具体来说就是： 首先通过spring.factories文件，找到一个注册相关的Configuration配置类，这个配置类里面定义了三个Bean对象。创建第三个Bean对象时，需要第一个、第二个Bean对象作为参数传进去。第一个Bean对象里面就有真正进行服务注册的register()方法，并且第一个Bean对象会赋值给第三个Bean对象中的serviceRegistry属性，在第三个Bean对象的父类会实现Spring的监听器方法。所以在Spring容器启动时会发布监听事件，从而触发执行Nacos注册逻辑。
(3)Nacos客户端通过gRPC方式发起服务实例注册
nacos-client会提供接口给Nacos客户端调用来进行服务实例注册。
在NacosNamingService提供的服务注册接口registerInstance()中，会调用NamingClientProxyDelegate的registerService()方法来注册服务。此时会先调用NamingClientProxyDelegate的getExecuteClientProxy()方法，来判断要注册的服务实例是否为临时实例来获取gRPC代理还是HTTP代理。如果注册的是临时实例，则使用gRPC方式注册，否则用HTTP方式注册，然后再调用NamingGrpcClientProxy的registerService()方法注册服务。
在NamingGrpcClientProxy的registerService()方法中，则会调用NamingGrpcClientProxy的doRegisterService()方法执行注册。此时先根据要注册的服务信息创建一个InstanceRequest请求参数对象，然后调用NamingGrpcClientProxy的requestToServer()方法发出请求，也就是通过调用RpcClient的request()方法向Nacos服务端发出gRPC请求。
至于RpcClient的request()方法的底层实现，则是通过一个本地存根代理类grpcFutureServiceStub调用gRPC的接口，来实现向Nacos服务端发起RPC调用的。
//Nacos Naming Service. @SuppressWarnings("PMD.ServiceOrDaoClassShouldEndWithImplRule") public class NacosNamingService implements NamingService { ... private NamingClientProxy clientProxy; private void init(Properties properties) throws NacosException { ... //初始化clientProxy属性为NamingClientProxyDelegate对象实例 this.clientProxy = new NamingClientProxyDelegate(this.namespace, serviceInfoHolder, properties, changeNotifier); } //register a instance to service with specified instance properties. @Override public void registerInstance(String serviceName, String groupName, Instance instance) throws NacosException { NamingUtils.checkInstanceIsLegal(instance); //调用NamingClientProxy的注册方法registerService()，其实就是NamingClientProxyDelegate.registerService()方法 clientProxy.registerService(serviceName, groupName, instance); } ... } //Delegate of naming client proxy. public class NamingClientProxyDelegate implements NamingClientProxy { private final NamingHttpClientProxy httpClientProxy; private final NamingGrpcClientProxy grpcClientProxy; public NamingClientProxyDelegate(String namespace, ServiceInfoHolder serviceInfoHolder, Properties properties, InstancesChangeNotifier changeNotifier) throws NacosException { ... this.httpClientProxy = new NamingHttpClientProxy(namespace, securityProxy, serverListManager, properties, serviceInfoHolder); this.grpcClientProxy = new NamingGrpcClientProxy(namespace, securityProxy, serverListManager, properties, serviceInfoHolder); } ... @Override public void registerService(String serviceName, String groupName, Instance instance) throws NacosException { getExecuteClientProxy(instance).registerService(serviceName, groupName, instance); } private NamingClientProxy getExecuteClientProxy(Instance instance) { return instance.isEphemeral() ? grpcClientProxy : httpClientProxy; } ... } //Naming grpc client proxy. public class NamingGrpcClientProxy extends AbstractNamingClientProxy { private final String namespaceId; private final String uuid; private final Long requestTimeout; private final RpcClient rpcClient; private final NamingGrpcRedoService redoService; public NamingGrpcClientProxy(String namespaceId, SecurityProxy securityProxy, ServerListFactory serverListFactory, Properties properties, ServiceInfoHolder serviceInfoHolder) throws NacosException { super(securityProxy); this.namespaceId = namespaceId; this.uuid = UUID.randomUUID().toString(); this.requestTimeout = Long.parseLong(properties.getProperty(CommonParams.NAMING_REQUEST_TIMEOUT, "-1")); Map<String, String> labels = new HashMap<String, String>(); labels.put(RemoteConstants.LABEL_SOURCE, RemoteConstants.LABEL_SOURCE_SDK); labels.put(RemoteConstants.LABEL_MODULE, RemoteConstants.LABEL_MODULE_NAMING); //通过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()方法建立与服务端的连接 rpcClient.start(); NotifyCenter.registerSubscriber(this); } ... //Register a instance to service with specified instance properties. @Override public void registerService(String serviceName, String groupName, Instance instance) throws NacosException { NAMING_LOGGER.info("[REGISTER-SERVICE] {} registering service {} with instance {}", namespaceId, serviceName, instance); redoService.cacheInstanceForRedo(serviceName, groupName, instance); //执行服务实例的注册 doRegisterService(serviceName, groupName, instance); } //Execute register operation. public void doRegisterService(String serviceName, String groupName, Instance instance) throws NacosException { //创建请求参数对象InstanceRequest InstanceRequest request = new InstanceRequest(namespaceId, serviceName, groupName, NamingRemoteConstants.REGISTER_INSTANCE, instance); //向服务端发起请求 requestToServer(request, Response.class); redoService.instanceRegistered(serviceName, groupName); } 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"); } ... } //abstract remote client to connect to server. @SuppressWarnings("PMD.AbstractClassShouldStartWithAbstractNamingRule") public abstract class RpcClient implements Closeable { protected volatile Connection currentConnection; ... public final void start() throws NacosException { ... // connect to server, try to connect to server sync RETRY_TIMES times, async starting if failed. Connection connectToServer = null; rpcClientStatus.set(RpcClientStatus.STARTING); int startUpRetryTimes = RETRY_TIMES; while (startUpRetryTimes > 0 && connectToServer == null) { try { startUpRetryTimes--; ServerInfo serverInfo = nextRpcServer(); LoggerUtils.printIfInfoEnabled(LOGGER, "[{}] Try to connect to server on start up, server: {}", name, serverInfo); //调用GrpcClient.connectToServer()方法建立和服务端的长连接 connectToServer = connectToServer(serverInfo); } catch (Throwable e) { LoggerUtils.printIfWarnEnabled(LOGGER, "[{}] Fail to connect to server on start up, error message = {}, start up retry times left: {}", name, e.getMessage(), startUpRetryTimes); } } if (connectToServer != null) { LoggerUtils.printIfInfoEnabled(LOGGER, "[{}] Success to connect to server [{}] on start up, connectionId = {}", name, connectToServer.serverInfo.getAddress(), connectToServer.getConnectionId()); //currentConnection其实就是一个用于客户端的GrpcConnection对象实例 this.currentConnection = connectToServer; rpcClientStatus.set(RpcClientStatus.RUNNING); eventLinkedBlockingQueue.offer(new ConnectionEvent(ConnectionEvent.CONNECTED)); } else { switchServerAsync(); } ... } //connect to server. public abstract Connection connectToServer(ServerInfo serverInfo) throws Exception; ... //send request. public Response request(Request request) throws NacosException { return request(request, DEFAULT_TIMEOUT_MILLS); } //send request. public Response request(Request request, long timeoutMills) throws NacosException { int retryTimes = 0; Response response; Exception exceptionThrow = null; long start = System.currentTimeMillis(); while (retryTimes < RETRY_TIMES && System.currentTimeMillis() < timeoutMills + start) { boolean waitReconnect = false; try { if (this.currentConnection == null || !isRunning()) { waitReconnect = true; throw new NacosException(NacosException.CLIENT_DISCONNECT, "Client not connected, current status:" + rpcClientStatus.get()); } //发起gRPC请求，调用GrpcConnection.request()方法 response = this.currentConnection.request(request, timeoutMills); if (response == null) { throw new NacosException(SERVER_ERROR, "Unknown Exception."); } if (response instanceof ErrorResponse) { if (response.getErrorCode() == NacosException.UN_REGISTER) { synchronized (this) { waitReconnect = true; if (rpcClientStatus.compareAndSet(RpcClientStatus.RUNNING, RpcClientStatus.UNHEALTHY)) { LoggerUtils.printIfErrorEnabled(LOGGER, "Connection is unregistered, switch server, connectionId = {}, request = {}", currentConnection.getConnectionId(), request.getClass().getSimpleName()); switchServerAsync(); } } } throw new NacosException(response.getErrorCode(), response.getMessage()); } //return response. lastActiveTimeStamp = System.currentTimeMillis(); return response; } catch (Exception e) { if (waitReconnect) { try { //wait client to reconnect. Thread.sleep(Math.min(100, timeoutMills / 3)); } catch (Exception exception) { //Do nothing. } } LoggerUtils.printIfErrorEnabled(LOGGER, "Send request fail, request = {}, retryTimes = {}, errorMessage = {}", request, retryTimes, e.getMessage()); exceptionThrow = e; } retryTimes++; } if (rpcClientStatus.compareAndSet(RpcClientStatus.RUNNING, RpcClientStatus.UNHEALTHY)) { switchServerAsyncOnRequestFail(); } if (exceptionThrow != null) { throw (exceptionThrow instanceof NacosException) ? (NacosException) exceptionThrow : new NacosException(SERVER_ERROR, exceptionThrow); } else { throw new NacosException(SERVER_ERROR, "Request fail, unknown Error"); } } ... } //gRPC connection. public class GrpcConnection extends Connection { //stub to send request. //grpcFutureServiceStub属性会在NamingGrpcClientProxy初始化 -> 调用RpcClient.start() -> GrpcClient.connectToServer()时， //通过GrpcConnection.setGrpcFutureServiceStub()方法进行设置 protected RequestGrpc.RequestFutureStub grpcFutureServiceStub; ... @Override public Response request(Request request, long timeouts) throws NacosException { Payload grpcRequest = GrpcUtils.convert(request); //调用gRPC提供的接口发起请求，属于io.grpc包下的内容了 ListenableFuture<Payload> requestFuture = grpcFutureServiceStub.request(grpcRequest); Payload grpcResponse; try { grpcResponse = requestFuture.get(timeouts, TimeUnit.MILLISECONDS); } catch (Exception e) { throw new NacosException(NacosException.SERVER_ERROR, e); } return (Response) GrpcUtils.parse(grpcResponse); } ... }(4)总结
3.服务端进行服务注册时的处理
(1)服务端处理客户端发起的服务注册请求
(2)服务端对客户端注册事件的处理源码
(3)总结
(1)服务端处理客户端发起的服务注册请求
客户端向服务端发起服务注册时，会先根据要注册的服务信息来创建一个InstanceRequest请求参数对象，再调用NamingGrpcClientProxy的requestToServer()方法向服务端发请求，也就是通过调用RpcClient的request()方法来向服务端发出gRPC请求。
InstanceRequestHandler的handle()方法就是用来处理服务注册请求的，该方法会继续调用InstanceRequestHandler的registerInstance()方法，根据客户端发过来的请求类型来选择是注册服务还是注销服务。
如果客户端发过来的请求类型是注册服务实例，则调用EphemeralClientOperationServiceImpl的registerInstance()方法，在该方法中：
一.首先会调用ServiceManager的getSingleton()方法
根据由请求信息创建的Service对象获取一个已注册的Service对象。
需要注意：在Nacos 1.x中，ServiceManager使用一个双层Map存放服务和命名空间。在Nacos 2.x中，ServiceManager则使用两个Map存放服务和命名空间。
在Nacos 1.x中的双层Map是： Map<String, Map<String, Service>>； 例如Map(namespace, Map(group::serviceName, Service))； 在Nacos 2.x中的两个Map是： ConcurrentHashMap<Service, Service>、 ConcurrentHashMap<String_namespace, Set<Service>>；二.然后调用ClientManagerDelegate的getClient()方法
根据请求参数中的connectionId来获取一个IpPortBasedClient对象。在执行ClientManagerDelegate的getClient()方法时，会先根据connectionId选出具体的ClientManager实现类，接着再调用比如EphemeralIpPortClientManager的getClient()方法，从EphemeralIpPortClientManager.clients属性中获取一个Client对象。clients属性是一个ConcurrentMap，key是请求参数中的connectionId，value是继承了实现Client接口的AbstractClient的IpPortBasedClient对象。
需要注意：Nacos中的gRPC底层是基于Netty实现的。当客户端和服务端建立长连接后，服务端会生成SocketChannel连接对象，这个SocketChannel连接对象就代表了客户端。Nacos会在这个SocketChannel连接对象的基础上，封装一个Client对象，并且生成一个connectionId将SocketChannel对象与Client对象关联起来。
所以当要进行服务注册的客户端和服务端建立好长连接后，服务端就会为客户端创建一个IpPortBasedClient对象，并将该对象存放在EphemeralIpPortClientManager.clients属性里。
三.接着调用ClientOperationService的getPublishInfo()方法
将请求中的instance实例信息封装为InstancePublishInfo对象。
四.然后调用IpPortBasedClient对象的addServiceInstance()方法
往IpPortBasedClient对象里添加Service对象 -> InstancePublishInfo对象。
由于IpPortBasedClient继承自实现了Client接口的AbstractClient抽象类，所以实际是调用AbstractClient的addServiceInstance()方法添加服务实例。
在AbstractClient抽象类中，有一个名为publishers的属性。它是一个ConcurrentHashMap，用于存储客户端服务注册请求中的Instance信息。就是记录该客户端提供的服务和服务实例，一个客户端可提供多个服务，当然这些信息已经被封装为InstancePublishInfo对象了。所以AbstractClient.publishers属性的key为已注册的Service，value是根据请求中的Instance实例信息封装的InstancePublishInfo对象。
在AbstractClient的addServiceInstance()方法中，首先往publishers放入这次要注册的Service对象和客户端Instance实例，然后发布客户端改变事件ClientChangedEvent，用来同步集群间的数据。
五.最后发布客户端注册服务实例事件和服务实例元数据事件
客户端注册服务实例事件是ClientRegisterServiceEvent，服务实例元数据事件是InstanceMetadataEvent。
//Instance request handler. @Component public class InstanceRequestHandler extends RequestHandler<InstanceRequest, InstanceResponse> { private final EphemeralClientOperationServiceImpl clientOperationService; public InstanceRequestHandler(EphemeralClientOperationServiceImpl clientOperationService) { this.clientOperationService = clientOperationService; } @Override @Secured(action = ActionTypes.WRITE) public InstanceResponse handle(InstanceRequest request, RequestMeta meta) throws NacosException { //根据请求信息创建一个Service对象，里面包含了：命名空间、分组名、服务名 Service service = Service.newService(request.getNamespace(), request.getGroupName(), request.getServiceName(), true); switch (request.getType()) { case NamingRemoteConstants.REGISTER_INSTANCE: //注册实例 return registerInstance(service, request, meta); case NamingRemoteConstants.DE_REGISTER_INSTANCE: //注销实例 return deregisterInstance(service, request, meta); default: throw new NacosException(NacosException.INVALID_PARAM, String.format("Unsupported request type %s", request.getType())); } } private InstanceResponse registerInstance(Service service, InstanceRequest request, RequestMeta meta) { //调用EphemeralClientOperationServiceImpl的注册方法registerInstance()，这里需要注意如下参数； //参数service：根据请求信息创建的一个Service对象，里面有命名空间、分组名、服务名 //参数request.getInstance()：这个参数就对应了客户端的实例对象，里面包含IP、端口等信息 //参数meta.getConnectionId()：这个参数很关键，它是连接ID clientOperationService.registerInstance(service, request.getInstance(), meta.getConnectionId()); return new InstanceResponse(NamingRemoteConstants.REGISTER_INSTANCE); } private InstanceResponse deregisterInstance(Service service, InstanceRequest request, RequestMeta meta) { //调用EphemeralClientOperationServiceImpl的注销方法deregisterInstance() clientOperationService.deregisterInstance(service, request.getInstance(), meta.getConnectionId()); return new InstanceResponse(NamingRemoteConstants.DE_REGISTER_INSTANCE); } } //Operation service for ephemeral clients and services. @Component("ephemeralClientOperationService") public class EphemeralClientOperationServiceImpl implements ClientOperationService { private final ClientManager clientManager; public EphemeralClientOperationServiceImpl(ClientManagerDelegate clientManager) { this.clientManager = clientManager; } @Override public void registerInstance(Service service, Instance instance, String clientId) { //1.从ServiceManager中根据由请求信息创建的Service对象获取一个已注册的Service对象 Service singleton = ServiceManager.getInstance().getSingleton(service); if (!singleton.isEphemeral()) { throw new NacosRuntimeException(NacosException.INVALID_PARAM, String.format("Current service %s is persistent service, can't register ephemeral instance.", singleton.getGroupedServiceName())); } //2.从ClientManagerDelegate中根据请求参数中的connectionId获取一个Client对象，即IpPortBasedClient对象 Client client = clientManager.getClient(clientId); if (!clientIsLegal(client, clientId)) { return; } //3.将请求中的instance实例信息封装为InstancePublishInfo对象 InstancePublishInfo instanceInfo = getPublishInfo(instance); //4.往Client对象里添加已注册的服务对象Service，调用的是IpPortBasedClient对象的父类AbstractClient的addServiceInstance()方法 client.addServiceInstance(singleton, instanceInfo); //设置IpPortBasedClient对象的lastUpdatedTime属性为最新时间 client.setLastUpdatedTime(); //5.发布客户端注册服务实例的事件 NotifyCenter.publishEvent(new ClientOperationEvent.ClientRegisterServiceEvent(singleton, clientId)); //5.发布服务实例元数据的事件 NotifyCenter.publishEvent(new MetadataEvent.InstanceMetadataEvent(singleton, instanceInfo.getMetadataId(), false)); } ... } public interface ClientOperationService { ... //get publish info. default InstancePublishInfo getPublishInfo(Instance instance) { InstancePublishInfo result = new InstancePublishInfo(instance.getIp(), instance.getPort()); Map<String, Object> extendDatum = result.getExtendDatum(); if (null != instance.getMetadata() && !instance.getMetadata().isEmpty()) { extendDatum.putAll(instance.getMetadata()); } if (StringUtils.isNotEmpty(instance.getInstanceId())) { extendDatum.put(Constants.CUSTOM_INSTANCE_ID, instance.getInstanceId()); } if (Constants.DEFAULT_INSTANCE_WEIGHT != instance.getWeight()) { extendDatum.put(Constants.PUBLISH_INSTANCE_WEIGHT, instance.getWeight()); } if (!instance.isEnabled()) { extendDatum.put(Constants.PUBLISH_INSTANCE_ENABLE, instance.isEnabled()); } String clusterName = StringUtils.isBlank(instance.getClusterName()) ? UtilsAndCommons.DEFAULT_CLUSTER_NAME : instance.getClusterName(); result.setHealthy(instance.isHealthy()); result.setCluster(clusterName); return result; } } //Nacos service manager for v2. public class ServiceManager { private static final ServiceManager INSTANCE = new ServiceManager(); //key是根据请求参数创建的Service对象，value是已经注册的Service对象 private final ConcurrentHashMap<Service, Service> singletonRepository; //key是命名空间，value是相同命名空间的已注册的Service对象集合 private final ConcurrentHashMap<String, Set<Service>> namespaceSingletonMaps; private ServiceManager() { singletonRepository = new ConcurrentHashMap<>(1 << 10); namespaceSingletonMaps = new ConcurrentHashMap<>(1 << 2); } public static ServiceManager getInstance() { return INSTANCE; } public Set<Service> getSingletons(String namespace) { return namespaceSingletonMaps.getOrDefault(namespace, new HashSet<>(1)); } //Get singleton service. Put to manager if no singleton. //@param service new service //@return if service is exist, return exist service, otherwise return new service public Service getSingleton(Service service) { //往singletonRepository这个ConcurrentHashMap中添加一个Service对象，如果存在就不添加 singletonRepository.putIfAbsent(service, service); //从这个ConcurrentHashMap中把已注册的Service对象取出来 Service result = singletonRepository.get(service); //将已注册的Service对象添加到namespaceSingletonMaps中 //由于namespaceSingletonMaps会按命名空间对已注册的Service对象进行分类 //所以namespaceSingletonMaps的key是命名空间，value是相同命名空间的已注册的Service对象集合 namespaceSingletonMaps.computeIfAbsent(result.getNamespace(), (namespace) -> new ConcurrentHashSet<>()); namespaceSingletonMaps.get(result.getNamespace()).add(result); return result; } ... } //Client manager delegate. @Component("clientManager") public class ClientManagerDelegate implements ClientManager { private final ConnectionBasedClientManager connectionBasedClientManager; private final EphemeralIpPortClientManager ephemeralIpPortClientManager; private final PersistentIpPortClientManager persistentIpPortClientManager; public ClientManagerDelegate(ConnectionBasedClientManager connectionBasedClientManager, EphemeralIpPortClientManager ephemeralIpPortClientManager, PersistentIpPortClientManager persistentIpPortClientManager) { this.connectionBasedClientManager = connectionBasedClientManager; this.ephemeralIpPortClientManager = ephemeralIpPortClientManager; this.persistentIpPortClientManager = persistentIpPortClientManager; } ... @Override public Client getClient(String clientId) { //通过请求参数中的connectionId获取一个Client对象，比如调用EphemeralIpPortClientManager.getClient()方法 return getClientManagerById(clientId).getClient(clientId); } private ClientManager getClientManagerById(String clientId) { if (isConnectionBasedClient(clientId)) { return connectionBasedClientManager; } return clientId.endsWith(ClientConstants.PERSISTENT_SUFFIX) ? persistentIpPortClientManager : ephemeralIpPortClientManager; } ... } //The manager of {@code IpPortBasedClient} and ephemeral. @Component("ephemeralIpPortClientManager") public class EphemeralIpPortClientManager implements ClientManager { //key是请求参数中的connectionId即clientId，value是一个继承了实现Client接口的AbstractClient的IpPortBasedClient对象 private final ConcurrentMap<String, IpPortBasedClient> clients = new ConcurrentHashMap<>(); private final DistroMapper distroMapper; private final ClientFactory<IpPortBasedClient> clientFactory; public EphemeralIpPortClientManager(DistroMapper distroMapper, SwitchDomain switchDomain) { this.distroMapper = distroMapper; GlobalExecutor.scheduleExpiredClientCleaner(new ExpiredClientCleaner(this, switchDomain), 0, Constants.DEFAULT_HEART_BEAT_INTERVAL, TimeUnit.MILLISECONDS); clientFactory = ClientFactoryHolder.getInstance().findClientFactory(ClientConstants.EPHEMERAL_IP_PORT); } @Override public boolean clientConnected(String clientId, ClientAttributes attributes) { return clientConnected(clientFactory.newClient(clientId, attributes)); } @Override public boolean clientConnected(final Client client) { clients.computeIfAbsent(client.getClientId(), s -> { Loggers.SRV_LOG.info("Client connection {} connect", client.getClientId()); IpPortBasedClient ipPortBasedClient = (IpPortBasedClient) client; ipPortBasedClient.init(); return ipPortBasedClient; }); return true; } @Override public boolean syncClientConnected(String clientId, ClientAttributes attributes) { return clientConnected(clientFactory.newSyncedClient(clientId, attributes)); } @Override public Client getClient(String clientId) { //客户端和服务端建立长连接时，就会通过EphemeralIpPortClientManager.clientConnected()方法将clientId放入到clients return clients.get(clientId); } ... } public class IpPortBasedClient extends AbstractClient { ... ... } 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); ... @Override public boolean addServiceInstance(Service service, InstancePublishInfo instancePublishInfo) { //服务注册时，如果是第一次put进去Service对象，会返回null if (null == publishers.put(service, instancePublishInfo)) { //监视器记录 MetricsMonitor.incrementInstanceCount(); } //发布客户端改变事件ClientChangedEvent，用于处理集群间的数据同步 NotifyCenter.publishEvent(new ClientEvent.ClientChangedEvent(this)); Loggers.SRV_LOG.info("Client change for service {}, {}", service, getClientId()); return true; } ... }(2)服务端对客户端注册事件的处理源码
一.服务端处理客户端的服务注册请求时的主要工作
往三个ConcurrentHashMap里放入内容 + 发布三个事件。
三个ConcurrentHashMap分别是：第一个是ServiceManager中的，key是根据请求参数创建的Service对象，value是已经注册的Service对象。第二个是ServiceManager中的，key是命名空间，value是相同命名空间的已注册的Service对象集合。第三个是AbstractClient中的，key是注册的Service服务对象，value是根据请求中的Instance实例封装的InstancePublishInfo对象。
三个事件分别是：第一个是在AbstractClient的addServiceInstance()方法中，发布的ClientChangedEvent客户端改变事件。第二个是在EphemeralClientOperationService的registerInstance()方法中，发布的ClientRegisterServiceEvent客户端注册事件。第三个是在EphemeralClientOperationService的registerInstance()方法中，发布的InstanceMetadataEvent服务实例元数据事件。
二.服务端对客户端注册事件ClientRegisterServiceEvent的处理
当执行EphemeralClientOperationService的registerInstance()方法，发布一个ClientRegisterServiceEvent客户端注册事件时，便会触发执行ClientServiceIndexesManager的onEvent()方法，然后执行ClientServiceIndexesManager的handleClientOperation()方法，最终调用ClientServiceIndexesManager的addPublisherIndexes()方法。
其中addPublisherIndexes()方法会把clientId放入到publisherIndexes中。publisherIndexes是一个ConcurrentMap，它的key是要注册的服务实例所属的服务Service对象，它的value是某服务Service对象下的所有clientId即connectionId。由于connectionId代表了一个Client对象，也就是一个客户端，所以publisherIndexes的value可理解为服务Service下的所有客户端实例。
//Client and service index manager. @Component public class ClientServiceIndexesManager extends SmartSubscriber { //注册表(服务提供者)，一个Service服务对象，对应多个服务实例的clientId private final ConcurrentMap<Service, Set<String>> publisherIndexes = new ConcurrentHashMap<>(); ... //可以处理客户端注册事件ClientRegisterServiceEvent @Override public 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) { //处理客户端注册事件ClientRegisterServiceEvent addPublisherIndexes(service, clientId); } else if (event instanceof ClientOperationEvent.ClientDeregisterServiceEvent) { //处理客户端注销事件ClientDeregisterServiceEvent removePublisherIndexes(service, clientId); } else if (event instanceof ClientOperationEvent.ClientSubscribeServiceEvent) { //处理客户端订阅服务事件ClientSubscribeServiceEvent addSubscriberIndexes(service, clientId); } else if (event instanceof ClientOperationEvent.ClientUnsubscribeServiceEvent) { //处理客户端取消订阅事件ClientUnsubscribeServiceEvent removeSubscriberIndexes(service, clientId); } } private void addPublisherIndexes(Service service, String clientId) { //判断注册表是否存在该Service，不存在则创建一个空的ConcurrentHashSet publisherIndexes.computeIfAbsent(service, (key) -> new ConcurrentHashSet<>()); //把clientId放入到对应的Service中 publisherIndexes.get(service).add(clientId); //发布服务改变事件 NotifyCenter.publishEvent(new ServiceEvent.ServiceChangedEvent(service, true)); } ... }(3)总结
服务端处理服务实例注册时，会使用多个Map来存储微服务实例的信息。在注册表也就是ClientServiceIndexesManager.publisherIndexes属性中，只是简单记录每个Service服务对象下包含的clientId字符串集合。通过clientId可以在clients属性中获取到IpPortBasedClient对象，IpPortBasedClient的父类AbstractClient会存储对应的Instance实例信息，所以这样的注册表是可记录每个Service服务包含的所有Instance实例的。
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 secs public 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; } @Override public List<NacosServer> getInitialListOfServers() { return getServers(); } @Override public 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; } @Override public 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); } @Override public List<Instance> selectInstances(String serviceName, String groupName, List<String> clusters, boolean healthy, boolean subscribe) throws NacosException { ServiceInfo serviceInfo; String clusterString = StringUtils.join(clusters, ","); //判断是否需要订阅，默认为true if (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()); //获取key String 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); } ... @Override public 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); } ... @Override public 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) { //生成一个serverKey String serviceKey = ServiceInfo.getKey(NamingUtils.getGroupedName(serviceName, groupName), clusters); //判断当前serviceKey是否已经开启了对应的定时任务，如果已经开启就不开启了 if (futureMap.get(serviceKey) != null) { return; } //加一把同步锁，避免并发冲突 synchronized (futureMap) { //加锁之后再进行一次判断，双重检测Double Check if (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 empty private 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); } @Override public 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. //计算下一次定时任务执行的时间，这个时间默认是6s delayTime = serviceObj.getCacheMillis() * DEFAULT_UPDATE_CACHE_TIME_MULTIPLE; //请求成功后，重置错误次数为0 resetFailCount(); } 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-serivce Service service = Service.newService(namespaceId, groupName, serviceName, true); //构建要订阅Service服务的订阅者，对应的是order-service Subscriber 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-serivce Set<Instance> result = new HashSet<>(); Set<String> clusters = new HashSet<>(); //each = clientId，调用ClientServiceIndexesManager.getAllClientsRegisteredService()方法获取Service服务对象的所有clientId for (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客户端注册的服务实例信息InstancePublishInfo return 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服务对象，对应多个服务实例的clientId private 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，不存在则创建一个空的ConcurrentHashSet publisherIndexes.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<>(); ... @Override public Client getClient(String clientId) { //客户端和服务端建立长连接时，就会通过EphemeralIpPortClientManager.clientConnected()方法将clientId放入到clients return 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); ... @Override public 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 @Override public void subscribeService(Service service, Subscriber subscriber, String clientId) { //传入的service是要查询的Service对象，比如stock-service Service 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<>(); ... @Override public Client getClient(String clientId) { //客户端和服务端建立长连接时，就会通过EphemeralIpPortClientManager.clientConnected()方法将clientId放入到clients return 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); ... @Override public 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服务对象，对应多个服务实例的clientId private final ConcurrentMap<Service, Set<String>> publisherIndexes = new ConcurrentHashMap<>(); //订阅者列表(服务消费者)，一个Service服务对象，对应多个订阅者的clientId private final ConcurrentMap<Service, Set<String>> subscriberIndexes = new ConcurrentHashMap<>(); ... //可以处理客户端注册事件ClientRegisterServiceEvent @Override public 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) { //处理客户端注册事件ClientRegisterServiceEvent addPublisherIndexes(service, clientId); } else if (event instanceof ClientOperationEvent.ClientDeregisterServiceEvent) { //处理客户端注销事件ClientDeregisterServiceEvent removePublisherIndexes(service, clientId); } else if (event instanceof ClientOperationEvent.ClientSubscribeServiceEvent) { //处理客户端订阅服务事件ClientSubscribeServiceEvent addSubscriberIndexes(service, clientId); } else if (event instanceof ClientOperationEvent.ClientUnsubscribeServiceEvent) { //处理客户端取消订阅事件ClientUnsubscribeServiceEvent removeSubscriberIndexes(service, clientId); } } private void addSubscriberIndexes(Service service, String clientId) { //传入的service是要查询的Service对象stock-service，clientId是订阅者order-service对应的客户端连接对象ID subscriberIndexes.computeIfAbsent(service, (key) -> new ConcurrentHashSet<>()); //Fix #5404, Only first time add need notify event. 只有第一次添加时需要发布通知事件 if (subscriberIndexes.get(service).add(clientId)) { //发布服务订阅事件ServiceSubscribedEvent NotifyCenter.publishEvent(new ServiceEvent.ServiceSubscribedEvent(service, clientId)); } } ... }(3)总结
![洛谷 P11345 [KTSC 2023 R2] 基地简化 题解](http://www.itfaba.com/wp-content/themes/kemi/timthumb.php?src=http://www.itfaba.com/wp-content/themes/kemi/img/random/3.jpg&w=218&h=124&zc=1)
