我们前面提到了，Nacos2.X版本中出现了很多的事件驱动架构的业务处理，对于很多逻辑，例如客户端注册/下线；服务改变；服务订阅等等都是发起一个Event，然后对应的有个Handler处理器进行处理

整体结构分析

我们可以来看一下之前我们已经了解过的几个Event类对象，其实可以发现，都是继承自Event这个事件父类

同样的，我们再去看一下所有的Handler所在类，也可以发现，有些继承自SmartSubscriber，并且SmartSubscriber之上还有个顶级接口Subscriber，就是订阅者，在这里定义了顶级抽象方法

public abstract List<Class<? extends Event>> subscribeTypes();

该方法的主要作用就是返回该订阅者感兴趣的Event类型的List集合

除此之外在顶级接口中Subscriber中还有一个onEvent()方法，该方法就是当对应的Event来了之后进行处理回调的方法

在每个子类的构造方法上，还会调用一个registerSubscriber方法，该方法是NotifyCenter中的静态方法

NotifyCenter.registerSubscriber(this);
NotifyCenter.registerSubscriber(this, NamingEventPublisherFactory.getInstance());

根据如上的机构，其实我们完全可以自己基于Nacos提供的事件驱动架构，自己写一些事件和订阅者进行运行

如下，我们先写一个TestEvent测试事件类

import com.alibaba.nacos.common.notify.Event;
​
public class TestEvent extends Event {
}
​

然后写一个对应的订阅者TestSubscriber，实现subscribeTypes和onEvent方法，在构造方法中将该订阅者注册上去

​
import com.alibaba.nacos.common.notify.Event;
import com.alibaba.nacos.common.notify.NotifyCenter;
import com.alibaba.nacos.common.notify.listener.SmartSubscriber;
import org.springframework.stereotype.Component;
​
import java.util.LinkedList;
import java.util.List;
​
@Component
public class TestSubscriber extends SmartSubscriber {
​
    public TestSubscriber() {
        NotifyCenter.registerSubscriber(this);
    }
​
    @Override
    public List<Class<? extends Event>> subscribeTypes() {
        List<Class<? extends Event>> result = new LinkedList<>();
        result.add(TestEvent.class);
        return result;
    }
​
    @Override
    public void onEvent(Event event) {
        System.out.println("TestSubscriber onEvent");
    }
}
​

我们写一个测试类，来测试一下上面的代码，可以看到，调用完publishEvent之后，对应的订阅者的onEvent方法也成功被调用且打印了输出

源码分析

事件发布

上面我们了解一下事件驱动架构的大致结构，我们现在就可以开始分析，我们先从事件的发布这块入手

我们知道，发布事件用的是NotifyCenter.publishEvent这个方法，我们来看下这个方法的源码

NotifyCenter.publishEvent(new TestEvent());
​
​
public static boolean publishEvent(final Event event) {
    try {
        // 跟主线任务
        return publishEvent(event.getClass(), event);
    } catch (Throwable ex) {
        LOGGER.error("There was an exception to the message publishing : ", ex);
        return false;
    }
}
​
​
private static boolean publishEvent(final Class<? extends Event> eventType, final Event event) {
    if (ClassUtils.isAssignableFrom(SlowEvent.class, eventType)) {
        return INSTANCE.sharePublisher.publish(event);
    }
    
    // 根据Event的类型的到事件类型Topic
    final String topic = ClassUtils.getCanonicalName(eventType);
    // 根据发布事件类型获取 EventPublisher 对象，该对象中会包含订阅者信息
    // 订阅者怎么来的？ 从这里可以看到肯定是从一个Map中获取的，具体的在下文章节会讲到，publisherMap其实就是一个以Event全类名为key，EventPublisher对象(该对象成员属性subscribers会保存对应的订阅者类)为value的map，
    EventPublisher publisher = INSTANCE.publisherMap.get(topic);
    if (publisher != null) {
        // 跟主线任务
        return publisher.publish(event);
    }
    LOGGER.warn("There are no [{}] publishers for this event, please register", topic);
    return false;
}
​
​
@Override
public boolean publish(Event event) {
    checkIsStart();
    // 把事件放入到了一个 阻塞队列
    boolean success = this.queue.offer(event);
    // 如果没有放入成功则手动调用receiveEvent方法接收并通知Subscriber处理该事件
    if (!success) {
        LOGGER.warn("Unable to plug in due to interruption, synchronize sending time, event : {}", event);
        receiveEvent(event);
        return true;
    }
    return true;
}
​
​

可以看到，又是利用到了阻塞队列，自然我们就可以去找到队列中拿任务的逻辑

可以发现就在DefaultPublisher中，在这一段代码中，我们有看到 subscribers 一个 Set 集合，注意，这里并不是全部的订阅者，只是当前发布的事件对应的订阅者，并不会通知全部的订阅者哈

for (; ; ) {
    if (shutdown) {
        break;
    }
    // 从阻塞队列取数据
    final Event event = queue.take();
    // 处理事件
    receiveEvent(event);
    UPDATER.compareAndSet(this, lastEventSequence, Math.max(lastEventSequence, event.sequence()));
}
​
​
protected final ConcurrentHashSet<Subscriber> subscribers = new ConcurrentHashSet<>();
​
void receiveEvent(Event event) {
    final long currentEventSequence = event.sequence();
    
    if (!hasSubscriber()) {
        LOGGER.warn("[NotifyCenter] the {} is lost, because there is no subscriber.", event);
        return;
    }
    
    // 循环遍历通知发布事件对应的订阅者
    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);
    }
}
​

然后最后调用了 notifySubscriber(subscriber, event); 方法来通知订阅者，代码如下:

通过这段代码，我们能够知道，最后是使用线程异步的方式，来通知订阅者的。 事件发布流程我们就分析完了，核心关键还是利用了 阻塞队列 + 异步任务 来进行实现的，那么接下来我们一起来看看，订阅者注册的流程。

@Override
public void notifySubscriber(final Subscriber subscriber, final Event event) {
    
    LOGGER.debug("[NotifyCenter] the {} will received by {}", event, subscriber);
    
    // 创建一个任务，就是调用订阅者的 onEvent 方法
    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.registerSubscriber()

可以看到，首先这个就是调用subscribeTypes获取当前订阅者订阅的Event类型的List列表，然后遍历这个列表，调用INSTANCE.sharePublisher.addSubscriber封装保存或者调用addSubscriber这个方法(两种情况的底层都将内容保存到了subscribers这个Set中)

public TestSubscriber() {
    NotifyCenter.registerSubscriber(this);
}
​
public static void registerSubscriber(final Subscriber consumer) {
    registerSubscriber(consumer, DEFAULT_PUBLISHER_FACTORY);
}
​
public static void registerSubscriber(final Subscriber consumer, final EventPublisherFactory factory) {
    // If you want to listen to multiple events, you do it separately,
    // based on subclass's subscribeTypes method return list, it can register to publisher.
    if (consumer instanceof SmartSubscriber) {
        // 在这里会调用 subscribeTypes 方法，来获取我们需要监听的事件，然后进行遍历
        for (Class<? extends Event> subscribeType : ((SmartSubscriber) consumer).subscribeTypes()) {
            // For case, producer: defaultSharePublisher -> consumer: smartSubscriber.
            if (ClassUtils.isAssignableFrom(SlowEvent.class, subscribeType)) {
                INSTANCE.sharePublisher.addSubscriber(consumer, subscribeType);
            } else {
                // For case, producer: defaultPublisher -> consumer: subscriber.
                // 添加订阅者
                addSubscriber(consumer, subscribeType, factory);
            }
        }
        return;
    }
    
    // 这里代码省略
}
​
​

接下来我们继续看addSubscriber方法

private static void addSubscriber(final Subscriber consumer, Class<? extends Event> subscribeType,
        EventPublisherFactory factory) {
    
    final String topic = ClassUtils.getCanonicalName(subscribeType);
    synchronized (NotifyCenter.class) {
        // 这里很关键，会创建 EventPublisher 对象，一个事件会对应一个 EventPublisher 一个对象
        MapUtil.computeIfAbsent(INSTANCE.publisherMap, topic, factory, subscribeType, ringBufferSize);
    }
    // 获取事件对应的 EventPublisher 对象
    EventPublisher publisher = INSTANCE.publisherMap.get(topic);
    if (publisher instanceof ShardedEventPublisher) {
        ((ShardedEventPublisher) publisher).addSubscriber(consumer, subscribeType);
    } else {
        // 往 EventPublisher 对象添加订阅者信息
        publisher.addSubscriber(consumer);
    }
}
​

其中MapUtil.computeIfAbsent的逻辑如下，可以看到，在这里会利用传入的第三个参数，调用其apply方法，利用第四个和第五个参数创建出一个EventPublisher对象，这个对象内部会存储我们的订阅者信息，也就是 subscribers 属性

@NotThreadSafe
public static <K, C, V, T> V computeIfAbsent(Map<K, V> target, K key, BiFunction<C, T, V> mappingFunction, C param1,
        T param2) {
    
    Objects.requireNonNull(target, "target");
    Objects.requireNonNull(key, "key");
    Objects.requireNonNull(mappingFunction, "mappingFunction");
    Objects.requireNonNull(param1, "param1");
    Objects.requireNonNull(param2, "param2");
    
    V val = target.get(key);
    if (val == null) {
        V ret = mappingFunction.apply(param1, param2);
        target.put(key, ret);
        return ret;
    }
    return val;
}

在代码的最后，会调用 addSubscriber 方法往subscribers 属性中添加订阅者，完成注册。

总结

Nacos2.X中的事件驱动架构，整体基于Event(事件)-Subscriber(订阅者)-EventPublish(事件发布者)-NotifyCenter(事件中心)

订阅者通过注册中心进行注册，通过订阅者subscribeTypes()方法来获取其关注的Event事件类型的集合，对于非慢事件利用工厂创建对应的EventPublisher，并且将订阅者保存到EventPublisher 的 subscribers 集合，在publisherMap中保存事件-事件发布者的对应关系，EventPublisher 初始化时启动异步线程，监听阻塞队列，等待事件到来

通过事件中心发布事件的时候，根据事件的类型获得其全类名作为topic，利用topic从publisherMap中获取对应的EventPublisher，利用EventPublisher将任务写入到阻塞任务队列中