• Nacos源码详解

    因为Eureka的闭源，Nacos成为了现在Spring Cloud微服务注册中心的主流方案，那咱们废话不多说，直接开始读源码(基于2.2.0版本)。

    首先我们知道Nacos是基于Springboot实现自动注入的，老规矩我们来到spring-cloud-alibaba-nacos-discovery包下面的META-INF\spring.factories配置文件：

    1. org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
    2.   com.alibaba.cloud.nacos.discovery.NacosDiscoveryAutoConfiguration,\
    3.   com.alibaba.cloud.nacos.ribbon.RibbonNacosAutoConfiguration,\
    4.   com.alibaba.cloud.nacos.endpoint.NacosDiscoveryEndpointAutoConfiguration,\
    5.   com.alibaba.cloud.nacos.registry.NacosServiceRegistryAutoConfiguration,\
    6.   com.alibaba.cloud.nacos.discovery.NacosDiscoveryClientConfiguration,\
    7.   com.alibaba.cloud.nacos.discovery.reactive.NacosReactiveDiscoveryClientConfiguration,\
    8.   com.alibaba.cloud.nacos.discovery.configclient.NacosConfigServerAutoConfiguration
    9. org.springframework.cloud.bootstrap.BootstrapConfiguration=\
    10.   com.alibaba.cloud.nacos.discovery.configclient.NacosDiscoveryClientConfigServiceBootstrapConfiguration

    我们首先来到NacosDiscoveryAutoConfiguration

    1. @Configuration(proxyBeanMethods = false)
    2. @ConditionalOnDiscoveryEnabled
    3. @ConditionalOnNacosDiscoveryEnabled
    4. public class NacosDiscoveryAutoConfiguration {
    5.  
    6. 	@Bean
    7. 	@ConditionalOnMissingBean
    8. 	public NacosDiscoveryProperties nacosProperties() {
    9. 		return new NacosDiscoveryProperties();
    10. 	}
    11.  
    12. 	@Bean
    13. 	@ConditionalOnMissingBean
    14. 	public NacosServiceDiscovery nacosServiceDiscovery(
    15. 			NacosDiscoveryProperties discoveryProperties) {
    16. 		return new NacosServiceDiscovery(discoveryProperties);
    17. 	}
    18.  
    19. }

    这个自动配置类注入了两个Bean，第一个Bean是从配置文件中读spring.cloud.nacos.discovery前缀的配置项，然后是第二个Bean

    1. public class NacosServiceDiscovery {
    2.  
    3. 	private NacosDiscoveryProperties discoveryProperties;
    4.  
    5. 	public NacosServiceDiscovery(NacosDiscoveryProperties discoveryProperties) {
    6. 		this.discoveryProperties = discoveryProperties;
    7. 	}
    8.  
    9. 	/**
    10. 	 * Return all instances for the given service.
    11. 	 * @param serviceId id of service
    12. 	 * @return list of instances
    13. 	 * @throws NacosException nacosException
    14. 	 */
    15. 	public List getInstances(String serviceId) throws NacosException {
    16. 		String group = discoveryProperties.getGroup();
    17. 		List instances = discoveryProperties.namingServiceInstance()
    18. 				.selectInstances(serviceId, group, true);
    19. 		return hostToServiceInstanceList(instances, serviceId);
    20. 	}
    21.  
    22. 	/**
    23. 	 * Return the names of all services.
    24. 	 * @return list of service names
    25. 	 * @throws NacosException nacosException
    26. 	 */
    27. 	public List getServices() throws NacosException {
    28. 		String group = discoveryProperties.getGroup();
    29. 		ListView services = discoveryProperties.namingServiceInstance()
    30. 				.getServicesOfServer(1, Integer.MAX_VALUE, group);
    31. 		return services.getData();
    32. 	}
    33.  
    34. 	public static List hostToServiceInstanceList(
    35. 			List instances, String serviceId) {
    36. 		List result = new ArrayList<>(instances.size());
    37. 		for (Instance instance : instances) {
    38. 			ServiceInstance serviceInstance = hostToServiceInstance(instance, serviceId);
    39. 			if (serviceInstance != null) {
    40. 				result.add(serviceInstance);
    41. 			}
    42. 		}
    43. 		return result;
    44. 	}
    45.  
    46. 	public static ServiceInstance hostToServiceInstance(Instance instance,
    47. 			String serviceId) {
    48. 		if (instance == null || !instance.isEnabled() || !instance.isHealthy()) {
    49. 			return null;
    50. 		}
    51. 		NacosServiceInstance nacosServiceInstance = new NacosServiceInstance();
    52. 		nacosServiceInstance.setHost(instance.getIp());
    53. 		nacosServiceInstance.setPort(instance.getPort());
    54. 		nacosServiceInstance.setServiceId(serviceId);
    55.  
    56. 		Map metadata = new HashMap<>();
    57. 		metadata.put("nacos.instanceId", instance.getInstanceId());
    58. 		metadata.put("nacos.weight", instance.getWeight() + "");
    59. 		metadata.put("nacos.healthy", instance.isHealthy() + "");
    60. 		metadata.put("nacos.cluster", instance.getClusterName() + "");
    61. 		metadata.putAll(instance.getMetadata());
    62. 		nacosServiceInstance.setMetadata(metadata);
    63.  
    64. 		if (metadata.containsKey("secure")) {
    65. 			boolean secure = Boolean.parseBoolean(metadata.get("secure"));
    66. 			nacosServiceInstance.setSecure(secure);
    67. 		}
    68. 		return nacosServiceInstance;
    69. 	}

    根据代码和注释可以看出，这个Bean的核心功能是根据配置文件里Nacos的相关配置，获取实例和实例对象转化的一些功能，然后是NacosDiscoveryEndpointAutoConfiguration

    1. @Configuration(proxyBeanMethods = false)
    2. @ConditionalOnClass(Endpoint.class)
    3. @ConditionalOnNacosDiscoveryEnabled
    4. public class NacosDiscoveryEndpointAutoConfiguration {
    5.  
    6. 	@Bean
    7. 	@ConditionalOnMissingBean
    8. 	@ConditionalOnEnabledEndpoint
    9. 	public NacosDiscoveryEndpoint nacosDiscoveryEndpoint(
    10. 			NacosDiscoveryProperties nacosDiscoveryProperties) {
    11. 		return new NacosDiscoveryEndpoint(nacosDiscoveryProperties);
    12. 	}
    13.  
    14. 	@Bean
    15. 	@ConditionalOnEnabledHealthIndicator("nacos-discovery")
    16. 	public HealthIndicator nacosDiscoveryHealthIndicator(
    17. 			NacosDiscoveryProperties nacosDiscoveryProperties) {
    18. 		return new NacosDiscoveryHealthIndicator(
    19. 				nacosDiscoveryProperties.namingServiceInstance());
    20. 	}
    21.  
    22. }

    这个类注入了两个Bean，第一个Bean NacosDiscoveryEndpoint 的作用是作为Nacos服务发现的端点，获取当前客户端的所有注册的服务。

    第二个Bean NacosDiscoveryHealthIndicator的作用是通过向Nacos注册中心请求/operator/metrics 接口来确认健康状态。

    接下来这个自动配置类NacosServiceRegistryAutoConfiguration比较核心

    1. @Configuration(proxyBeanMethods = false)
    2. @EnableConfigurationProperties
    3. /*是否开启Nacos服务发现*/
    4. @ConditionalOnNacosDiscoveryEnabled
    5. /*是否开启自动注册，默认为True*/
    6. @ConditionalOnProperty(value = "spring.cloud.service-registry.auto-registration.enabled",
    7. 		matchIfMissing = true)
    8. /*在以下几个配置类装配完成后才进行装配*/
    9. @AutoConfigureAfter({ AutoServiceRegistrationConfiguration.class,
    10. 		AutoServiceRegistrationAutoConfiguration.class,
    11. 		NacosDiscoveryAutoConfiguration.class })
    12. public class NacosServiceRegistryAutoConfiguration {
    13.     /*读取配置文件*/
    14. 	@Bean
    15. 	public NacosServiceRegistry nacosServiceRegistry(
    16. 			NacosDiscoveryProperties nacosDiscoveryProperties) {
    17. 		return new NacosServiceRegistry(nacosDiscoveryProperties);
    18. 	}
    19.     /*Nacos服务注册Bean*/
    20. 	@Bean
    21. 	@ConditionalOnBean(AutoServiceRegistrationProperties.class)
    22. 	public NacosRegistration nacosRegistration(
    23. 			NacosDiscoveryProperties nacosDiscoveryProperties,
    24. 			ApplicationContext context) {
    25. 		return new NacosRegistration(nacosDiscoveryProperties, context);
    26. 	}
    27.     /*Nacos服务自动注册Bean*/
    28. 	@Bean
    29. 	@ConditionalOnBean(AutoServiceRegistrationProperties.class)
    30. 	public NacosAutoServiceRegistration nacosAutoServiceRegistration(
    31. 			NacosServiceRegistry registry,
    32. 			AutoServiceRegistrationProperties autoServiceRegistrationProperties,
    33. 			NacosRegistration registration) {
    34. 		return new NacosAutoServiceRegistration(registry,
    35. 				autoServiceRegistrationProperties, registration);
    36. 	}
    37.  
    38. }

    该配置类装配了三个Bean，我们重点关注下第三个同来实现服务自动注册的Bean NacosAutoServiceRegistration

    1. public class NacosAutoServiceRegistration
    2. 		extends AbstractAutoServiceRegistration {
    3.  
    4. 	private static final Logger log = LoggerFactory
    5. 			.getLogger(NacosAutoServiceRegistration.class);
    6.  
    7. 	private NacosRegistration registration;
    8.  
    9. 	public NacosAutoServiceRegistration(ServiceRegistry serviceRegistry,
    10. 			AutoServiceRegistrationProperties autoServiceRegistrationProperties,
    11. 			NacosRegistration registration) {
    12. 		super(serviceRegistry, autoServiceRegistrationProperties);
    13. 		this.registration = registration;
    14. 	}
    15.  
    16. 	@Deprecated
    17. 	public void setPort(int port) {
    18. 		getPort().set(port);
    19. 	}
    20.  
    21. 	@Override
    22. 	protected NacosRegistration getRegistration() {
    23. 		if (this.registration.getPort() < 0 && this.getPort().get() > 0) {
    24. 			this.registration.setPort(this.getPort().get());
    25. 		}
    26. 		Assert.isTrue(this.registration.getPort() > 0, "service.port has not been set");
    27. 		return this.registration;
    28. 	}
    29.  
    30. 	@Override
    31. 	protected NacosRegistration getManagementRegistration() {
    32. 		return null;
    33. 	}
    34.  
    35. 	@Override
    36. 	protected void register() {
    37. 		if (!this.registration.getNacosDiscoveryProperties().isRegisterEnabled()) {
    38. 			log.debug("Registration disabled.");
    39. 			return;
    40. 		}
    41. 		if (this.registration.getPort() < 0) {
    42. 			this.registration.setPort(getPort().get());
    43. 		}
    44. 		super.register();
    45. 	}
    46.  
    47. 	@Override
    48. 	protected void registerManagement() {
    49. 		if (!this.registration.getNacosDiscoveryProperties().isRegisterEnabled()) {
    50. 			return;
    51. 		}
    52. 		super.registerManagement();
    53.  
    54. 	}
    55.  
    56. 	@Override
    57. 	protected Object getConfiguration() {
    58. 		return this.registration.getNacosDiscoveryProperties();
    59. 	}
    60.  
    61. 	@Override
    62. 	protected boolean isEnabled() {
    63. 		return this.registration.getNacosDiscoveryProperties().isRegisterEnabled();
    64. 	}
    65.  
    66. 	@Override
    67. 	@SuppressWarnings("deprecation")
    68. 	protected String getAppName() {
    69. 		String appName = registration.getNacosDiscoveryProperties().getService();
    70. 		return StringUtils.isEmpty(appName) ? super.getAppName() : appName;
    71. 	}

    通过名字我们可以看出，这个Bean的核心是redister方法，该方法调用的是父抽象类AbstractAutoServiceRegistration中的实现，然后在父类中它又调用了接口ServiceRegistry的register方法，实际由NacosServiceRegistry实现，我们进入到NacosServiceRegistry实现的register方法

    1. 	public void register(Registration registration) {
    2.  
    3. 		if (StringUtils.isEmpty(registration.getServiceId())) {
    4. 			log.warn("No service to register for nacos client...");
    5. 			return;
    6. 		}
    7. 		String serviceId = registration.getServiceId();
    8. 		String group = nacosDiscoveryProperties.getGroup();
    9.  
    10. 		Instance instance = getNacosInstanceFromRegistration(registration);
    11.  
    12. 		try {
    13. 			namingService.registerInstance(serviceId, group, instance);
    14. 			log.info("nacos registry, {} {} {}:{} register finished", group, serviceId,
    15. 					instance.getIp(), instance.getPort());
    16. 		}
    17. 		catch (Exception e) {
    18. 			log.error("nacos registry, {} register failed...{},", serviceId,
    19. 					registration.toString(), e);
    20. 			// rethrow a RuntimeException if the registration is failed.
    21. 			// issue : https://github.com/alibaba/spring-cloud-alibaba/issues/1132
    22. 			rethrowRuntimeException(e);
    23. 		}
    24. 	}

    然后我们重点关注其中namingService.registerInstance方法，namingService是一个接口，由NacosNamingService实现，我们进入到方法实现

    1.     public void registerInstance(String serviceName, String groupName, Instance instance) throws NacosException {
    2.         //首先判断该实例是临时还是持久的，默认是临时的
    3.         if (instance.isEphemeral()) {
    4.             //新建心跳对象
    5.             BeatInfo beatInfo = new BeatInfo();
    6.             //设置serviceName
    7.             beatInfo.setServiceName(NamingUtils.getGroupedName(serviceName, groupName));
    8.             //设置IP
    9.             beatInfo.setIp(instance.getIp());
    10.             //设置端口号
    11.             beatInfo.setPort(instance.getPort());
    12.             //设置集群名
    13.             beatInfo.setCluster(instance.getClusterName());
    14.             //设置实例权重
    15.             beatInfo.setWeight(instance.getWeight());
    16.             beatInfo.setMetadata(instance.getMetadata());
    17.             beatInfo.setScheduled(false);
    18.             long instanceInterval = instance.getInstanceHeartBeatInterval();
    19.             //设置发送心跳时间间隔，默认5秒
    20.             beatInfo.setPeriod(instanceInterval == 0 ? DEFAULT_HEART_BEAT_INTERVAL : instanceInterval);
    21.             //放入定时线程池等待执行
    22.             beatReactor.addBeatInfo(NamingUtils.getGroupedName(serviceName, groupName), beatInfo);
    23.         }
    24.         //进行服务注册
    25.         serverProxy.registerService(NamingUtils.getGroupedName(serviceName, groupName), groupName, instance);
    26.     }

    该方法首先封装了一个心跳对象，并放入ScheduledThreadPoolExecutor线程池中，默认每5秒执行一次，然后调用注册中心代理serverProxy通过向注册中心/instance接口发送进行服务注册

    1.     public void registerService(String serviceName, String groupName, Instance instance) throws NacosException {
    2.  
    3.         NAMING_LOGGER.info("[REGISTER-SERVICE] {} registering service {} with instance: {}",
    4.             namespaceId, serviceName, instance);
    5.  
    6.         final Map params = new HashMap(9);
    7.         params.put(CommonParams.NAMESPACE_ID, namespaceId);
    8.         params.put(CommonParams.SERVICE_NAME, serviceName);
    9.         params.put(CommonParams.GROUP_NAME, groupName);
    10.         params.put(CommonParams.CLUSTER_NAME, instance.getClusterName());
    11.         params.put("ip", instance.getIp());
    12.         params.put("port", String.valueOf(instance.getPort()));
    13.         params.put("weight", String.valueOf(instance.getWeight()));
    14.         params.put("enable", String.valueOf(instance.isEnabled()));
    15.         params.put("healthy", String.valueOf(instance.isHealthy()));
    16.         params.put("ephemeral", String.valueOf(instance.isEphemeral()));
    17.         params.put("metadata", JSON.toJSONString(instance.getMetadata()));
    18.  
    19.         reqAPI(UtilAndComs.NACOS_URL_INSTANCE, params, HttpMethod.POST);
    20.  
    21.     }

    我们来到注册中心的/instance接口，重点关注InstanceController的register方法

    1.     @CanDistro
    2.     @PostMapping
    3.     @TpsControl(pointName = "NamingInstanceRegister", name = "HttpNamingInstanceRegister")
    4.     @Secured(action = ActionTypes.WRITE)
    5.     public String register(HttpServletRequest request) throws Exception {
    6.         
    7.         final String namespaceId = WebUtils.optional(request, CommonParams.NAMESPACE_ID,
    8.                 Constants.DEFAULT_NAMESPACE_ID);
    9.         final String serviceName = WebUtils.required(request, CommonParams.SERVICE_NAME);
    10.         NamingUtils.checkServiceNameFormat(serviceName);
    11.         
    12.         final Instance instance = HttpRequestInstanceBuilder.newBuilder()
    13.                 .setDefaultInstanceEphemeral(switchDomain.isDefaultInstanceEphemeral()).setRequest(request).build();
    14.         
    15.         getInstanceOperator().registerInstance(namespaceId, serviceName, instance);
    16.         NotifyCenter.publishEvent(new RegisterInstanceTraceEvent(System.currentTimeMillis(), "", false, namespaceId,
    17.                 NamingUtils.getGroupName(serviceName), NamingUtils.getServiceName(serviceName), instance.getIp(),
    18.                 instance.getPort()));
    19.         return "ok";
    20.     }

    重点关注getInstanceOperator().registerInstance(namespaceId, serviceName, instance)方法

    1.     /**
    2.      * This method creates {@code IpPortBasedClient} if it doesn't exist.
    3.      */
    4.     @Override
    5.     public void registerInstance(String namespaceId, String serviceName, Instance instance) throws NacosException {
    6.         NamingUtils.checkInstanceIsLegal(instance);
    7.         //判断是否临时
    8.         boolean ephemeral = instance.isEphemeral();
    9.         //获取客户端ID
    10.         String clientId = IpPortBasedClient.getClientId(instance.toInetAddr(), ephemeral);
    11.         //如果客户端不存在则新建
    12.         createIpPortClientIfAbsent(clientId);
    13.         //封装service对象
    14.         Service service = getService(namespaceId, serviceName, ephemeral);
    15.         //进行实例注册
    16.         clientOperationService.registerInstance(service, instance, clientId);
    17.     }

    clientOperationService为一个接口且有多个实现，我们查看其实现类找到EphemeralClientOperationServiceImpl中的registerInstance方法

    1.     public void registerInstance(Service service, Instance instance, String clientId) throws NacosException {
    2.         //检查实例是否合法
    3.         NamingUtils.checkInstanceIsLegal(instance);
    4.         //获取单例的service
    5.         Service singleton = ServiceManager.getInstance().getSingleton(service);
    6.         if (!singleton.isEphemeral()) {
    7.             throw new NacosRuntimeException(NacosException.INVALID_PARAM,
    8.                     String.format("Current service %s is persistent service, can't register ephemeral instance.",
    9.                             singleton.getGroupedServiceName()));
    10.         }
    11.         //获取客户端
    12.         Client client = clientManager.getClient(clientId);
    13.         if (!clientIsLegal(client, clientId)) {
    14.             return;
    15.         }
    16.         //封装实例发布信息
    17.         InstancePublishInfo instanceInfo = getPublishInfo(instance);
    18.         client.addServiceInstance(singleton, instanceInfo);
    19.         client.setLastUpdatedTime();
    20.         client.recalculateRevision();
    21.         //通知中心发布客户端服务注册事件
    22.         NotifyCenter.publishEvent(new ClientOperationEvent.ClientRegisterServiceEvent(singleton, clientId));
    23.         NotifyCenter
    24.                 .publishEvent(new MetadataEvent.InstanceMetadataEvent(singleton, instanceInfo.getMetadataId(), false));
    25.     }

    然后我们关注publishEvent

    1.     /**
    2.      * Request publisher publish event Publishers load lazily, calling publisher.
    3.      *
    4.      * @param eventType class Instances type of the event type.
    5.      * @param event     event instance.
    6.      */
    7.     private static boolean publishEvent(final Class eventType, final Event event) {
    8.         if (ClassUtils.isAssignableFrom(SlowEvent.class, eventType)) {
    9.             return INSTANCE.sharePublisher.publish(event);
    10.         }
    11.         
    12.         final String topic = ClassUtils.getCanonicalName(eventType);
    13.         
    14.         EventPublisher publisher = INSTANCE.publisherMap.get(topic);
    15.         if (publisher != null) {
    16.             return publisher.publish(event);
    17.         }
    18.         if (event.isPluginEvent()) {
    19.             return true;
    20.         }
    21.         LOGGER.warn("There are no [{}] publishers for this event, please register", topic);
    22.         return false;
    23.     }

    在DefaultPublisher中的publish方法，加入阻塞队列等待执行

    1.     @Override
    2.     public boolean publish(Event event) {
    3.         checkIsStart();
    4.         //加入阻塞队列尾部
    5.         boolean success = this.queue.offer(event);
    6.         if (!success) {
    7.             LOGGER.warn("Unable to plug in due to interruption, synchronize sending time, event : {}", event);
    8.             //接收事件
    9.             receiveEvent(event);
    10.             return true;
    11.         }
    12.         return true;
    13.     }
    14.     
    15.     void checkIsStart() {
    16.         if (!initialized) {
    17.             throw new IllegalStateException("Publisher does not start");
    18.         }
    19.     }
    20.     
    21.     @Override
    22.     public void shutdown() {
    23.         this.shutdown = true;
    24.         this.queue.clear();
    25.     }
    26.     
    27.     public boolean isInitialized() {
    28.         return initialized;
    29.     }
    30.     
    31.     /**
    32.      * Receive and notifySubscriber to process the event.
    33.      *
    34.      * @param event {@link Event}.
    35.      */
    36.     void receiveEvent(Event event) {
    37.         final long currentEventSequence = event.sequence();
    38.         //检查有无订阅者
    39.         if (!hasSubscriber()) {
    40.             LOGGER.warn("[NotifyCenter] the {} is lost, because there is no subscriber.", event);
    41.             return;
    42.         }
    43.         // Notification single event listener
    44.         for (Subscriber subscriber : subscribers) {
    45.             if (!subscriber.scopeMatches(event)) {
    46.                 continue;
    47.             }
    48.             
    49.             // Whether to ignore expiration events
    50.             if (subscriber.ignoreExpireEvent() && lastEventSequence > currentEventSequence) {
    51.                 LOGGER.debug("[NotifyCenter] the {} is unacceptable to this subscriber, because had expire",
    52.                         event.getClass());
    53.                 continue;
    54.             }
    55.             
    56.             // Because unifying smartSubscriber and subscriber, so here need to think of compatibility.
    57.             // Remove original judge part of codes.
    58.             //通知订阅者
    59.             notifySubscriber(subscriber, event);
    60.         }
    61.     }
    62.     
    63.     @Override
    64.     public void notifySubscriber(final Subscriber subscriber, final Event event) {
    65.         
    66.         LOGGER.debug("[NotifyCenter] the {} will received by {}", event, subscriber);
    67.         
    68.         final Runnable job = () -> subscriber.onEvent(event);
    69.         final Executor executor = subscriber.executor();
    70.         
    71.         if (executor != null) {
    72.             executor.execute(job);
    73.         } else {
    74.             try {
    75.                 //执行事件
    76.                 job.run();
    77.             } catch (Throwable e) {
    78.                 LOGGER.error("Event callback exception: ", e);
    79.             }
    80.         }
    81.     }

    然后我们通过ClientRegisterServiceEvent的其他引用找到了ClientServiceIndexesManager中的onEvent方法，发现ClientRegisterServiceEvent事件被封装成了ServiceChangedEvent并发布

    1.     @Override
    2.     public void onEvent(Event event) {
    3.         if (event instanceof ClientOperationEvent.ClientReleaseEvent) {
    4.             handleClientDisconnect((ClientOperationEvent.ClientReleaseEvent) event);
    5.         } else if (event instanceof ClientOperationEvent) {
    6.             handleClientOperation((ClientOperationEvent) event);
    7.         }
    8.     }
    9.     
    10.     private void handleClientDisconnect(ClientOperationEvent.ClientReleaseEvent event) {
    11.         Client client = event.getClient();
    12.         for (Service each : client.getAllSubscribeService()) {
    13.             removeSubscriberIndexes(each, client.getClientId());
    14.         }
    15.         DeregisterInstanceReason reason = event.isNative()
    16.                 ? DeregisterInstanceReason.NATIVE_DISCONNECTED : DeregisterInstanceReason.SYNCED_DISCONNECTED;
    17.         long currentTimeMillis = System.currentTimeMillis();
    18.         for (Service each : client.getAllPublishedService()) {
    19.             removePublisherIndexes(each, client.getClientId());
    20.             InstancePublishInfo instance = client.getInstancePublishInfo(each);
    21.             NotifyCenter.publishEvent(new DeregisterInstanceTraceEvent(currentTimeMillis,
    22.                     "", false, reason, each.getNamespace(), each.getGroup(), each.getName(),
    23.                     instance.getIp(), instance.getPort()));
    24.         }
    25.     }
    26.     
    27.     private void handleClientOperation(ClientOperationEvent event) {
    28.         Service service = event.getService();
    29.         String clientId = event.getClientId();
    30.         if (event instanceof ClientOperationEvent.ClientRegisterServiceEvent) {
    31.             addPublisherIndexes(service, clientId);
    32.         } else if (event instanceof ClientOperationEvent.ClientDeregisterServiceEvent) {
    33.             removePublisherIndexes(service, clientId);
    34.         } else if (event instanceof ClientOperationEvent.ClientSubscribeServiceEvent) {
    35.             addSubscriberIndexes(service, clientId);
    36.         } else if (event instanceof ClientOperationEvent.ClientUnsubscribeServiceEvent) {
    37.             removeSubscriberIndexes(service, clientId);
    38.         }
    39.     }
    40.     
    41.     private void addPublisherIndexes(Service service, String clientId) {
    42.         publisherIndexes.computeIfAbsent(service, key -> new ConcurrentHashSet<>());
    43.         publisherIndexes.get(service).add(clientId);
    44.         NotifyCenter.publishEvent(new ServiceEvent.ServiceChangedEvent(service, true));
    45.     }

    而Nacos注册中心是通过JRaftServer的NacosStateMachine中RequestProcessor对象去执行onApply方法，而RequestProcessor为抽象类，其实现类为InstanceMetadataProcessor，我们来到其中的onApply方法

    1.     @Override
    2.     public Response onApply(WriteRequest request) {
    3.         readLock.lock();
    4.         try {
    5.             MetadataOperation op = serializer.deserialize(request.getData().toByteArray(), processType);
    6.             switch (DataOperation.valueOf(request.getOperation())) {
    7.                 case ADD:
    8.                 case CHANGE:
    9.                     updateInstanceMetadata(op);
    10.                     break;
    11.                 case DELETE:
    12.                     deleteInstanceMetadata(op);
    13.                     break;
    14.                 default:
    15.                     return Response.newBuilder().setSuccess(false)
    16.                             .setErrMsg("Unsupported operation " + request.getOperation()).build();
    17.             }
    18.             return Response.newBuilder().setSuccess(true).build();
    19.         } catch (Exception e) {
    20.             Loggers.RAFT.error("onApply {} instance metadata operation failed. ", request.getOperation(), e);
    21.             String errorMessage = null == e.getMessage() ? e.getClass().getName() : e.getMessage();
    22.             return Response.newBuilder().setSuccess(false).setErrMsg(errorMessage).build();
    23.         } finally {
    24.             readLock.unlock();
    25.         }
    26.     }
    27.     
    28.     private void updateInstanceMetadata(MetadataOperation op) {
    29.         Service service = Service.newService(op.getNamespace(), op.getGroup(), op.getServiceName());
    30.         service = ServiceManager.getInstance().getSingleton(service);
    31.         namingMetadataManager.updateInstanceMetadata(service, op.getTag(), op.getMetadata());
    32.         NotifyCenter.publishEvent(new ServiceEvent.ServiceChangedEvent(service, true));
    33.     }
    34.     
    35.     private void deleteInstanceMetadata(MetadataOperation op) {
    36.         Service service = Service.newService(op.getNamespace(), op.getGroup(), op.getServiceName());
    37.         service = ServiceManager.getInstance().getSingleton(service);
    38.         namingMetadataManager.removeInstanceMetadata(service, op.getTag());
    39.     }

    核心在NamingMetadataManager的updateInstanceMetadata中

    1.       private ConcurrentMap> instanceMetadataMap;
    2.   
    3. /**
    4.      * Update instance metadata.
    5.      *
    6.      * @param service          service
    7.      * @param metadataId       instance metadata id
    8.      * @param instanceMetadata new instance metadata
    9.      */
    10.     public void updateInstanceMetadata(Service service, String metadataId, InstanceMetadata instanceMetadata) {
    11.         if (!instanceMetadataMap.containsKey(service)) {
    12.             instanceMetadataMap.putIfAbsent(service, new ConcurrentHashMap<>(INITIAL_CAPACITY));
    13.         }
    14.         instanceMetadataMap.get(service).put(metadataId, instanceMetadata);
    15.     }

    其存放客户端实例的对象是一个两层的ConcurrentHashMap，第一层的Map key为service对象，第二层的Map Key为具体的实例Id，value为实例对象。

  • 相关阅读:
    使用爬虫批量下载图片链接并去重
    SMTP发送邮件时抱No appropriate protocol错误分析和解决方案
    centos rpm方式安装jenkins
    如何使用DBeaver连接Hive
    golang 函数式编程库samber/mo使用： Future
    Cache学习（1）：常见的程序运行模型&多级Cache存储结构
    如何将AI智能分析与视频监控平台EasyCVR相融合构建监狱安防体系
    centos7篇---安装nvidia-docker
    区域气象-大气化学在线耦合模式（WRF/Chem）在大气环境领域实践技术应用
    java数据结构与算法刷题-----LeetCode104：二叉树的最大深度
  • 原文地址：https://blog.csdn.net/qq_40992849/article/details/133203411