Spring Cloud Alibaba - Nacos源码分析(三)

目录

一、Nacos客户端服务订阅的事件机制

1、监听事件的注册

2、ServiceInfo处理

serviceInfoHolder.processServiceInfo

---

一、Nacos客户端服务订阅的事件机制

Nacos客户端订阅的核心流程：Nacos客户端通过一个定时任务，每6秒从注册中心获取实例列表，当发现实例发生变化时，发布变更事件，订阅者进行业务处理，然后更新内存中和本地的缓存中的实例。

        在第一步调用subscribe方法时，会订阅一个EventListener事件。而在定时任务UpdateTask定时获取实例列表之后，会调用ServiceInfoHolder.processServiceInfo方法对ServiceInfo进行本地处理，这其中就包括和事件处理。

1、监听事件的注册

在subscribe方法中，通过了下面的源码进行了监听事件的注册：

public class NacosNamingService implements NamingService {
...@Overridepublic void subscribe(String serviceName, String groupName, List<String> clusters, EventListener listener)throws NacosException {if (null == listener) {return;}String clusterString = StringUtils.join(clusters, ",");changeNotifier.registerListener(groupName, serviceName, clusterString, listener);clientProxy.subscribe(serviceName, groupName, clusterString);}

        在这其中我们主要要关注的就是changeNotifier.registerListener，此监听就是进行具体事件注册逻辑的，我们来看一下源码：
        可以看出，事件的注册便是将EventListener存储在InstancesChangeNotifier的listenerMap属性当中了。同时这里的数据结构为ConcurrentHashMap，key为服务实例的信息的拼接，value为监听事件的集合。

public class InstancesChangeNotifier extends Subscriber<InstancesChangeEvent> {private final String eventScope;private final Map<String, ConcurrentHashSet<EventListener>> listenerMap = new ConcurrentHashMap<>();
...public void registerListener(String groupName, String serviceName, String clusters, EventListener listener) {String key = ServiceInfo.getKey(NamingUtils.getGroupedName(serviceName, groupName), clusters);ConcurrentHashSet<EventListener> eventListeners = listenerMap.computeIfAbsent(key, keyInner -> new ConcurrentHashSet<>());eventListeners.add(listener);}

2、ServiceInfo处理

        上面的源码中已经完成了事件的注册，现在就来追溯触发事件的来源，UpdateTask中获取到最新的实例会进行本地化处理，部分源码如下：

// ServiceInfoUpdateService>UpdateTask>run()
ServiceInfo serviceObj = serviceInfoHolder.getServiceInfoMap().get(serviceKey);
if (serviceObj == null) {serviceObj = namingClientProxy.queryInstancesOfService(serviceName, groupName, clusters, 0, false);// 本地缓存处理serviceInfoHolder.processServiceInfo(serviceObj);lastRefTime = serviceObj.getLastRefTime();return;
}

serviceInfoHolder.processServiceInfo

这个逻辑简单来说：判断新的ServiceInfo数据是否正确，是否发生了变化。如果数据格式正确，且发生变化，那就发布一个InstancesChangeEvent事件，同时将ServiceInfo写入本地缓存。

public class ServiceInfoHolder implements Closeable {
...public ServiceInfo processServiceInfo(ServiceInfo serviceInfo) {String serviceKey = serviceInfo.getKey();if (serviceKey == null) {return null;}ServiceInfo oldService = serviceInfoMap.get(serviceInfo.getKey());if (isEmptyOrErrorPush(serviceInfo)) {//empty or error push, just ignorereturn oldService;}// 缓存服务信息serviceInfoMap.put(serviceInfo.getKey(), serviceInfo);// 判断注册的实例信息是否已变更boolean changed = isChangedServiceInfo(oldService, serviceInfo);if (StringUtils.isBlank(serviceInfo.getJsonFromServer())) {serviceInfo.setJsonFromServer(JacksonUtils.toJson(serviceInfo));}// 监控服务监控缓存Map的大小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(notifierEventScope, serviceInfo.getName(), serviceInfo.getGroupName(),serviceInfo.getClusters(), serviceInfo.getHosts()));// 记录Service本地文件DiskCache.write(serviceInfo, cacheDir);}return serviceInfo;}

分析到这里我们发现其实这个重点应该在服务信息变更之后，发布的InstancesChangeEvent事件，这个事件是NotifyCenter进行发布的，我们来追踪一下源码

Spring Cloud Alibaba - Nacos

干我们这行，啥时候懈怠，就意味着长进的停止，长进的停止就意味着被淘汰，只能往前冲，直到凤凰涅槃的一天！