ICode9
精准搜索请尝试： 精确搜索
首页
编程语言>
数据库
系统相关
互联网
其他分享
Python
Java
JavaScript
android
PHP
首页 > 编程语言> 文章详细

Nacos源码之客户端服务订阅事件机制剖析

2022-05-21 01:02:56  阅读：193  来源： 互联网

标签：订阅 serviceInfo Nacos EventPublisher 源码 NotifyCenter 事件 event 客户端


Nacos客户端服务订阅的事件机制剖析

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

image

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

监听事件的注册

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

@Override
public 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，此监听就是进行具体事件注册逻辑的，我们来看一下源码：

public void registerListener(String groupName, String serviceName, String clusters, EventListener listener) {
 String key = ServiceInfo.getKey(NamingUtils.getGroupedName(serviceName, groupName), clusters);
 ConcurrentHashSet<EventListener> eventListeners = listenerMap.get(key);
 if (eventListeners == null) {
     synchronized (lock) {
         eventListeners = listenerMap.get(key);
         if (eventListeners == null) {
             eventListeners = new ConcurrentHashSet<EventListener>();
             //将EventListener缓存到listenerMap
             listenerMap.put(key, eventListeners);
         }
     }
 }
 eventListeners.add(listener);
}

可以看出事件的注册便是将EventListener存储在InstancesChangeNotifier的listenerMap属性中。同时这里的数据结构为ConcurrentHashMap，key为服务实例信息的拼接，value为监听事件的集合。

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写入本地缓存。

image

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 ignore
     return 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(serviceInfo.getName(), serviceInfo.getGroupName(),
                                                        serviceInfo.getClusters(), serviceInfo.getHosts()));
     // 记录Service本地文件
     DiskCache.write(serviceInfo, cacheDir);
 }
 return serviceInfo;
}

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

事件追踪

NotifyCenter通知中心的核心流程如下：

image

NotifyCenter中进行事件发布，发布的核心逻辑是：

  1. 根据InstancesChangeEvent事件类型，获得对应的CanonicalName
  2. 将CanonicalName作为key，从NotifyCenter.publisherMap中获取对应的事件发布者(EventPublisher)
  3. EventPublisher将InstancesChangeEvent事件进行发布

核心代码如下：

private static boolean publishEvent(final Class<? extends Event> eventType, final Event event) {
    if (ClassUtils.isAssignableFrom(SlowEvent.class, eventType)) {
        return INSTANCE.sharePublisher.publish(event);
    }

    // 根据InstancesChangeEvent事件类型，获得对应的CanonicalName
    final String topic = ClassUtils.getCanonicalName(eventType);

    // 将CanonicalName作为Key，从NotifyCenter#publisherMap中获取对应的事件发布者（EventPublisher）
    EventPublisher publisher = INSTANCE.publisherMap.get(topic);
    if (publisher != null) {
        // 事件发布者publisher发布事件（InstancesChangeEvent）
        return publisher.publish(event);
    }
    LOGGER.warn("There are no [{}] publishers for this event, please register", topic);
    return false;
}

INSTANCE为NotifyCenter的单例实现。那么这里的publisherMap中key(CanonicalName)和value(EventPublisher)之间的关系是在NacosNamingService实例化时调用init初始化方法中进行绑定的

// Publisher的注册过程在于建立InstancesChangeEvent.class与EventPublisher的关系。
NotifyCenter.registerToPublisher(InstancesChangeEvent.class, 16384);

这里再继续跟踪registerToPublisher方法就会发现默认采用了DEFAULT_PUBLISHER_FACTORY（默认发布者工厂）来进行构建，我们再继续跟踪会发现，在NotifyCenter中静态代码块会发现DEFAULT_PUBLISHER_FACTORY默认构建的EventPublisher为DefaultPublisher。

//NotifyCenter
public static EventPublisher registerToPublisher(final Class<? extends Event> eventType, final int queueMaxSize) {
    return registerToPublisher(eventType, DEFAULT_PUBLISHER_FACTORY, queueMaxSize);
}
--------------------------------------------------------------------------------------------
//NotifyCenter>static中部分代码
DEFAULT_PUBLISHER_FACTORY = (cls, buffer) -> {
    try {
        EventPublisher publisher = clazz.newInstance();
        publisher.init(cls, buffer);
        return publisher;
    } catch (Throwable ex) {
        LOGGER.error("Service class newInstance has error : ", ex);
        throw new NacosRuntimeException(SERVER_ERROR, ex);
    }
};

所以我们得出结论NotifyCenter中它维护了事件名称和事件发布者的关系，而默认的事件发布者为DefaultPublisher。

DefaultPublisher的事件发布

默认事件发布者的源码，查看以后我们会发现它继承自Thread，也就是说它是一个线程类，同时，它又实现了EventPublisher，也就是发布者

public class DefaultPublisher extends Thread implements EventPublisher

来看它的init初始化方法，从这里我们可以看出当DefaultPublisher被初始化时，是以守护线程的方式运作的，其中还初始化了一个阻塞队列。

@Overridepublic void init(Class<? extends Event> type, int bufferSize) { // 守护线程 setDaemon(true); // 设置线程名字 setName("nacos.publisher-" + type.getName()); this.eventType = type; this.queueMaxSize = bufferSize; // 阻塞队列初始化 this.queue = new ArrayBlockingQueue<>(bufferSize); start();}

最后调用了start()方法：在这其中调用了super.start()启动线程

@Overridepublic synchronized void start() { if (!initialized) {     // start just called once     super.start();     if (queueMaxSize == -1) {         queueMaxSize = ringBufferSize;     }     initialized = true; }}

run()方法调用openEventHandler()方法

@Overridepublic void run() { openEventHandler();}void openEventHandler() { try {     // This variable is defined to resolve the problem which message overstock in the queue.     int waitTimes = 60;     // To ensure that messages are not lost, enable EventHandler when     // waiting for the first Subscriber to register     // 死循环，线程启动最大延时60秒，这个主要是为了解决消息积压的问题。     for (; ; ) {         if (shutdown || hasSubscriber() || waitTimes <= 0) {             break;         }         ThreadUtils.sleep(1000L);         waitTimes--;     }     // 死循环不断的从队列中取出Event，并通知订阅者Subscriber执行Event     for (; ; ) {         if (shutdown) {             break;         }         // 从队列中取出Event         final Event event = queue.take();         receiveEvent(event);         UPDATER.compareAndSet(this, lastEventSequence, Math.max(lastEventSequence, event.sequence()));     } } catch (Throwable ex) {     LOGGER.error("Event listener exception : ", ex); }}

这里写了两个死循环，第一个死循环可以理解为延时效果，也就是说线程启动时最大延时60秒，在这60秒中每隔1秒判断一下当前线程是否关闭，是否有订阅者，是否超过60秒。如果满足一个条件，就可以提前跳出死循环。而第二个死循环才是真正的业务逻辑处理，会从阻塞队列中取出一个事件，然后通过receiveEvent方法进行执行。

队列中的事件哪里来的？其实就是DefaultPublisher的发布事件方法被调用了publish往阻塞队列中存入事件，如果存入失败，会直接调用receiveEvent。可以理解为，如果向队列中存入失败，则立即执行，不走队列了。

@Overridepublic boolean publish(Event event) { checkIsStart(); // 向队列中插入事件元素 boolean success = this.queue.offer(event); // 判断是否成功插入 if (!success) {     LOGGER.warn("Unable to plug in due to interruption, synchronize sending time, event : {}", event);     // 失败直接执行     receiveEvent(event);     return true; } return true;}

receiveEvent方法的实现：这里其实就是遍历DefaultPublisher的subscribers（订阅者集合），然后执行通知订阅者的方法。

void receiveEvent(Event event) { final long currentEventSequence = event.sequence(); if (!hasSubscriber()) {     LOGGER.warn("[NotifyCenter] the {} is lost, because there is no subscriber.", event);     return; } // Notification single event listener // 通知订阅者执行Event for (Subscriber subscriber : subscribers) {     // Whether to ignore expiration events     if (subscriber.ignoreExpireEvent() && lastEventSequence > currentEventSequence) {         LOGGER.debug("[NotifyCenter] the {} is unacceptable to this subscriber, because had expire",                      event.getClass());         continue;     }     // Because unifying smartSubscriber and subscriber, so here need to think of compatibility.     // Remove original judge part of codes.     notifySubscriber(subscriber, event); }}

subscribers中订阅者是在NacosNamingService的init方法中：

// 将Subscribe注册到PublisherNotifyCenter.registerSubscriber(changeNotifier);

registerSubscriber方法最终会调用NotifyCenter的addSubscriber方法：核心逻辑就是将订阅事件、发布者、订阅者三者进行绑定。而发布者与事件通过Map进行维护、发布者与订阅者通过关联关系进行维护。

private static void addSubscriber(final Subscriber consumer, Class<? extends Event> subscribeType,                               EventPublisherFactory factory) { final String topic = ClassUtils.getCanonicalName(subscribeType); synchronized (NotifyCenter.class) {     // MapUtils.computeIfAbsent is a unsafe method.     MapUtil.computeIfAbsent(INSTANCE.publisherMap, topic, factory, subscribeType, ringBufferSize); } // 获取事件对应的Publisher EventPublisher publisher = INSTANCE.publisherMap.get(topic); if (publisher instanceof ShardedEventPublisher) {     ((ShardedEventPublisher) publisher).addSubscriber(consumer, subscribeType); } else {     // 添加到subscribers集合     publisher.addSubscriber(consumer); }}

关系都已经梳理明确了，事件也有了，最后我们看一下DefaulePublisher中的notifySubscriber方法，这里就是真正的订阅者执行事件了。

@Overridepublic void notifySubscriber(final Subscriber subscriber, final Event event) { LOGGER.debug("[NotifyCenter] the {} will received by {}", event, subscriber);	//执行订阅者事件 final Runnable job = () -> subscriber.onEvent(event); // 执行者 final Executor executor = subscriber.executor(); if (executor != null) {     executor.execute(job); } else {     try {         job.run();     } catch (Throwable e) {         LOGGER.error("Event callback exception: ", e);     } }}

总结

整体服务订阅的事件机制还是比较复杂的，因为用到了事件的形式，逻辑比较绕，并且其中还有守护线程，死循环，阻塞队列等。

需要重点理解NotifyCenter对事件发布者、事件订阅者和事件之间关系的维护，而这一关系的维护的入口就位于NacosNamingService的init方法当中。

核心流程梳理

ServiceInfoHolder中通过NotifyCenter发布了InstancesChangeEvent事件

NotifyCenter中进行事件发布，发布的核心逻辑是：

  • 根据InstancesChangeEvent事件类型，获得对应的CanonicalName
  • 将CanonicalName作为Key，从NotifyCenter.publisherMap中获取对应的事件发布者（EventPublisher）
  • EventPublisher将InstancesChangeEvent事件进行发布

InstancesChangeEvent事件发布：

  • 通过EventPublisher的实现类DefaultPublisher进行InstancesChangeEvent事件发布
  • DefaultPublisher本身以守护线程的方式运作，在执行业务逻辑前，先判断该线程是否启动
  • 如果启动，则将事件添加到BlockingQueue中，队列默认大小为16384
  • 添加到BlockingQueue成功，则整个发布过程完成
  • 如果添加失败，则直接调用DefaultPublisher.receiveEvent方法，接收事件并通知订阅者
  • 通知订阅者时创建一个Runnable对象，执行订阅者的Event
  • Event事件便是执行订阅时传入的事件

如果添加到BlockingQueue成功，则走另外一个业务逻辑：

  • DefaultPublisher初始化时会创建一个阻塞（BlockingQueue）队列，并标记线程启动
  • DefaultPublisher本身是一个Thread，当执行super.start方法时，会调用它的run方法
  • run方法的核心业务逻辑是通过openEventHandler方法处理的
  • openEventHandler方法通过两个for循环，从阻塞队列中获取时间信息
  • 第一个for循环用于让线程启动时在60s内检查执行条件
  • 第二个for循环为死循环，从阻塞队列中获取Event，并调用DefaultPublisher#receiveEvent方法，接收事件并通知订阅者
  • Event事件便是执行订阅时传入的事件

image

标签：订阅,serviceInfo,Nacos,EventPublisher,源码,NotifyCenter,事件,event,客户端
来源： https://www.cnblogs.com/ZT-666/p/16294227.html

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

关于我们 | 联系我们 | 留言反馈

专注分享技术，共同学习，共同进步。侵权联系[81616952@qq.com]

Copyright (C)ICode9.com, All Rights Reserved.

ICode9版权所有