• 微信公众号:美女很有趣。 工作之余,放松一下,关注即送10G+美女照片!

Nacos源码分析-事件发布机制

开发技术 开发技术 3小时前 1次浏览

温馨提示:
本文内容基于个人学习Nacos 2.0.1版本代码总结而来,因个人理解差异,不保证完全正确。如有理解错误之处欢迎各位拍砖指正,相互学习;转载请注明出处。

Nacos的服务注册、服务变更等功能都是通过事件发布来通知的,搞清楚事件发布订阅的机制,有利于理解业务的流程走向。本文将浅显的分析Nacos中的事件发布订阅实现。

事件(Event)

常规事件(Event)

package com.alibaba.nacos.common.notify;

public abstract class Event implements Serializable {
    
    private static final AtomicLong SEQUENCE = new AtomicLong(0);
    
    private final long sequence = SEQUENCE.getAndIncrement();
    
    /**
     * Event sequence number, which can be used to handle the sequence of events.
     *
     * @return sequence num, It's best to make sure it's monotone.
     */
    public long sequence() {
        return sequence;
    }
}

在事件抽象类中定义了一个事件的序列号,它是自增的。用于区分事件执行的前后顺序。它是由DefaultPublisher来处理。

慢事件(SlowEvent)

之所以称之为慢事件,可能因为所有的事件都共享同一个队列吧。

package com.alibaba.nacos.common.notify;

/**
 * This event share one event-queue.
 * @author <a href="mailto:liaochuntao@live.com">liaochuntao</a>
 * @author zongtanghu
 */
@SuppressWarnings("PMD.AbstractClassShouldStartWithAbstractNamingRule")
public abstract class SlowEvent extends Event {
    
    @Override
    public long sequence() {
        return 0;
    }
}

提示:
SlowEvent可以共享一个事件队列,也就是一个发布者可以同时管理多个事件的发布(区别于DefaultPublisher只能管理一个事件)。

订阅者(Subscriber)

单事件订阅者

这里的单事件订阅者指的是当前的订阅者只能订阅一种类型的事件。

package com.alibaba.nacos.common.notify.listener;

/**
 * An abstract subscriber class for subscriber interface.
 * @author <a href="mailto:liaochuntao@live.com">liaochuntao</a>
 * @author zongtanghu
 */
@SuppressWarnings("PMD.AbstractClassShouldStartWithAbstractNamingRule")
public abstract class Subscriber<T extends Event> {
    
    /**
     * Event callback.
     * 事件处理入口,由对应的事件发布器调用
     * @param event {@link Event}
     */
    public abstract void onEvent(T event);
    
    /**
     * Type of this subscriber's subscription.
     * 订阅的事件类型
     * @return Class which extends {@link Event}
     */
    public abstract Class<? extends Event> subscribeType();
    
    /**
     * It is up to the listener to determine whether the callback is asynchronous or synchronous.
     * 线程执行器,由具体的实现类来决定是异步还是同步调用
     * @return {@link Executor}
     */
    public Executor executor() {
        return null;
    }
    
    /**
     * Whether to ignore expired events.
     * 是否忽略过期事件
     * @return default value is {@link Boolean#FALSE}
     */
    public boolean ignoreExpireEvent() {
        return false;
    }
}

这是默认的订阅者对象,默认情况下一个订阅者只能订阅一个类型的事件。

多事件订阅者

package com.alibaba.nacos.common.notify.listener;


/**
 * Subscribers to multiple events can be listened to.
 *
 * @author <a href="mailto:liaochuntao@live.com">liaochuntao</a>
 * @author zongtanghu
 */
@SuppressWarnings("PMD.AbstractClassShouldStartWithAbstractNamingRule")
public abstract class SmartSubscriber extends Subscriber {
    
    /**
     * Returns which event type are smartsubscriber interested in.
     * 区别于父类,这里支持多个事件类型
     * @return The interestd event types.
     */
    public abstract List<Class<? extends Event>> subscribeTypes();
    
    @Override
    public final Class<? extends Event> subscribeType() {
		// 采用final修饰,禁止使用单一事件属性
        return null;
    }
    
    @Override
    public final boolean ignoreExpireEvent() {
        return false;
    }
}

提示
SmartSubscriber和Subscriber的区别是一个可以订阅多个事件,一个只能订阅一个事件,处理它们的发布者也不同。

发布者(Publisher)

发布者指的是Nacos中的事件发布者,顶级接口为EventPublisher。

package com.alibaba.nacos.common.notify;

/**
 * Event publisher.
 *
 * @author <a href="mailto:liaochuntao@live.com">liaochuntao</a>
 * @author zongtanghu
 */
public interface EventPublisher extends Closeable {
    
    /**
     * Initializes the event publisher.
     * 初始化事件发布者
     * @param type       {@link Event >}
     * @param bufferSize Message staging queue size
     */
    void init(Class<? extends Event> type, int bufferSize);
    
    /**
     * The number of currently staged events.
     * 当前暂存的事件数量
     * @return event size
     */
    long currentEventSize();
    
    /**
     * Add listener.
     * 添加订阅者
     * @param subscriber {@link Subscriber}
     */
    void addSubscriber(Subscriber subscriber);
    
    /**
     * Remove listener.
     * 移除订阅者
     * @param subscriber {@link Subscriber}
     */
    void removeSubscriber(Subscriber subscriber);
    
    /**
     * publish event.
     * 发布事件
     * @param event {@link Event}
     * @return publish event is success
     */
    boolean publish(Event event);
    
    /**
     * Notify listener.
     * 通知订阅者
     * @param subscriber {@link Subscriber}
     * @param event      {@link Event}
     */
    void notifySubscriber(Subscriber subscriber, Event event);
    
}

发布者的主要功能就是新增订阅者、通知订阅者,目前有两种类型的发布者分别是DefaultPublisher和DefaultSharePublisher。

单事件发布者(DefaultPublisher)

一个发布者实例只能处理一种类型的事件。

Nacos源码分析-事件发布机制

public class DefaultPublisher extends Thread implements EventPublisher {
	
	// 发布者是否初始化完毕
	private volatile boolean initialized = false;
	// 是否关闭了发布者
	private volatile boolean shutdown = false;
	// 事件的类型
	private Class<? extends Event> eventType;
	// 订阅者列表
	protected final ConcurrentHashSet<Subscriber> subscribers = new ConcurrentHashSet<Subscriber>();
	// 队列最大容量
	private int queueMaxSize = -1;
	// 队列类型
	private BlockingQueue<Event> queue;
	// 最后一个事件的序列号
	protected volatile Long lastEventSequence = -1L;
	// 事件序列号更新对象,用于更新原子属性lastEventSequence
	private static final AtomicReferenceFieldUpdater<DefaultPublisher, Long> UPDATER = AtomicReferenceFieldUpdater.newUpdater(DefaultPublisher.class, Long.class, "lastEventSequence");
}

发布者的初始化

public 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<Event>(bufferSize);
	start();
}

在初始化方法中,将其设置为了守护线程,意味着它将持续运行(它需要持续监控内部的事件队列),传入的type属性为当前发布者需要处理的事件类型,设置当前线程的名称以事件类型为区分,它将会以多个线程的形式存在,每个线程代表一种事件类型的发布者,后面初始化了队列的长度。最后调用启动方法完成当前线程的启动。

发布者线程启动

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

直接调用了Thread的start方法开启守护线程,并设置初始化状态为true。根据java线程的启动方式,调用start方法之后start方法是会调用run方法的。

public 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
		for (; ; ) {
			// 线程终止条件判断
			if (shutdown || hasSubscriber() || waitTimes <= 0) {
				break;
			}
			// 线程休眠1秒
			ThreadUtils.sleep(1000L);
			// 等待次数减1
			waitTimes--;
		}
		
		for (; ; ) {
			// 线程终止条件判断
			if (shutdown) {
				break;
			}
			// 从队列取出事件
			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);
	}
}

在run方法中调用了openEventHandler()方法。那发布者的实际工作原理就存在于这个方法内部。在首次启动的时候会等待1分钟,然后再进行消息消费。

接收并发布事件

这里的接收事件指的是接收通知中心发过来的事件,发布给订阅者。

void receiveEvent(Event event) {
	// 获取当前事件的序列号,它是自增的
	final long currentEventSequence = event.sequence();
	
	// 通知所有订阅了该事件的订阅者
	// Notification single event listener
	for (Subscriber subscriber : subscribers) {
		// 判断订阅者是否忽略事件过期,判断当前事件是否被处理过(lastEventSequence初始化的值为-1,而Event的sequence初始化的值为0)
		// 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);
	}
}

public void notifySubscriber(final Subscriber subscriber, final Event event) {
	
	LOGGER.debug("[NotifyCenter] the {} will received by {}", event, subscriber);
	
	// 为每个订阅者创建一个Runnable对象
	final Runnable job = () -> subscriber.onEvent(event);
	// 使用订阅者的线程执行器
	final Executor executor = subscriber.executor();
	// 若订阅者没有自己的执行器,则直接执行run方法启动订阅者消费线程
	if (executor != null) {
		executor.execute(job);
	} else {
		try {
			job.run();
		} catch (Throwable e) {
			LOGGER.error("Event callback exception: ", e);
		}
	}
}

外部调用发布事件

前面的发布事件是指从队列内部获取事件并通知订阅者,这里的发布事件区别在于它是开放给外部调用者,接收统一通知中心的事件并放入队列中的。

public 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;
}

在放入队列成功的时候直接返回,若放入队列失败,则是直接同步发送事件给订阅者,不经过队列。这里的同步我认为的是从调用者到发布者调用订阅者之间是同步的,若队列可用,则是调用者到入队列就完成了本次调用,不需要等待循环通知订阅者。使用队列解耦无疑会提升通知中心的工作效率。

总体来说就是一个发布者内部维护一个BlockingQueue,在实现上使用了ArrayBlockingQueue,它是一个有界阻塞队列,元素先进先出。并且使用非公平模式提升性能,意味着等待消费的订阅者执行顺序将得不到保障(业务需求没有这种顺序性要求)。同时也维护了一个订阅者集合(他们都订阅了同一个事件类型),在死循环中不断从ArrayBlockingQueue中获取数据来循环通知每一个订阅者,也就是调用订阅者的onEvent()方法。

多事件发布者(DefaultSharePublisher)

用于发布SlowEvent事件并通知所有订阅了该事件的订阅者。

public class DefaultSharePublisher extends DefaultPublisher {
	// 用于保存事件类型为SlowEvent的订阅者,一个事件类型对应多个订阅者
	private final Map<Class<? extends SlowEvent>, Set<Subscriber>> subMappings = new ConcurrentHashMap<Class<? extends SlowEvent>, Set<Subscriber>>();
    // 可重入锁
    private final Lock lock = new ReentrantLock();
}

它继承了DefaultPublisher,意味着它将拥有其所有的特性。从subMappings属性来看,这个发布器是支持多个SlowEvent事件的。DefaultSharePublisher重载了DefaultPublisher的addSubscriber()和removeSubscriber()方法,用于处理多事件类型的情形。

添加订阅者:

public void addSubscriber(Subscriber subscriber, Class<? extends Event> subscribeType) {
	
	// 将事件类型转换为当前发布者支持的类型
	// Actually, do a classification based on the slowEvent type.
	Class<? extends SlowEvent> subSlowEventType = (Class<? extends SlowEvent>) subscribeType;
	// 添加到父类的订阅者列表中,为何要添加呢?因为它需要使用父类的队列消费逻辑
	// For adding to parent class attributes synchronization.
	subscribers.add(subscriber);
	// 为多个操作加锁
	lock.lock();
	try {
		// 首先从事件订阅列表里面获取当前事件对应的订阅者集合
		Set<Subscriber> sets = subMappings.get(subSlowEventType);
		// 若没有订阅者,则新增当前订阅者
		if (sets == null) {
			Set<Subscriber> newSet = new ConcurrentHashSet<Subscriber>();
			newSet.add(subscriber);
			subMappings.put(subSlowEventType, newSet);
			return;
		}
		// 若当前事件订阅者列表不为空,则插入,因为使用的是Set集合因此可以避免重复数据
		sets.add(subscriber);
	} finally {
		// 别忘了解锁
		lock.unlock();
	}
}

提示:
Set newSet = new ConcurrentHashSet(); 它这里实际上使用的是自己实现的ConcurrentHashSet,它内部使用了ConcurrentHashMap来实现存储。
在ConcurrentHashSet.add()方法的实现上,它以当前插入的Subscriber对象为key,以一个Boolean值占位:map.putIfAbsent(o, Boolean.TRUE)。

事件类型和订阅者的存储状态为:
EventType1 -> {Subscriber1, Subscriber2, Subscriber3…}
EventType2 -> {Subscriber1, Subscriber2, Subscriber3…}
EventType3 -> {Subscriber1, Subscriber2, Subscriber3…}
感兴趣的可以自己查阅一下源码

移除订阅者

public void removeSubscriber(Subscriber subscriber, Class<? extends Event> subscribeType) {
	// 转换类型
	// Actually, do a classification based on the slowEvent type.
	Class<? extends SlowEvent> subSlowEventType = (Class<? extends SlowEvent>) subscribeType;
	// 先移除父类中的订阅者
	// For removing to parent class attributes synchronization.
	subscribers.remove(subscriber);
	// 加锁
	lock.lock();
	try {
		// 移除指定事件的指定订阅者
		Set<Subscriber> sets = subMappings.get(subSlowEventType);
		
		if (sets != null) {
			sets.remove(subscriber);
		}
	} finally {
		// 解锁
		lock.unlock();
	}
}

接收事件

@Override
public void receiveEvent(Event event) {
	// 获取当前事件的序列号
	final long currentEventSequence = event.sequence();
	// 获取事件的类型,转换为当前发布器支持的事件
	// get subscriber set based on the slow EventType.
	final Class<? extends SlowEvent> slowEventType = (Class<? extends SlowEvent>) event.getClass();
	
	// 获取当前事件的订阅者列表
	// Get for Map, the algorithm is O(1).
	Set<Subscriber> subscribers = subMappings.get(slowEventType);
	if (null == subscribers) {
		LOGGER.debug("[NotifyCenter] No subscribers for slow event {}", slowEventType.getName());
		return;
	}
	
	// 循环通知所有订阅者
	// Notification single event subscriber
	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;
		}
		// 通知逻辑和父类是共用的
		// Notify single subscriber for slow event.
		notifySubscriber(subscriber, event);
	}
}

提示:
DefaultPublisher是一个发布器只负责发布一个事件,并通知订阅了这个事件的所有订阅者;DefaultSharePublisher则是一个发布器可以发布多个事件,并通知订阅了这个事件的所有订阅者。

通知中心(NotifyCenter)

NotifyCenter 在Nacos中主要用于注册发布者、调用发布者发布事件、为发布者注册订阅者、为指定的事件增加指定的订阅者等操作。可以说它完全接管了订阅者、发布者和事件他们的组合过程。直接调用通知中心的相关方法即可实现事件发布订阅者注册等功能。

初始化信息

package com.alibaba.nacos.common.notify;

public class NotifyCenter {
	
    /**
     * 单事件发布者内部的事件队列初始容量
     */
    public static int ringBufferSize = 16384;

    /**
     * 多事件发布者内部的事件队列初始容量
     */
    public static int shareBufferSize = 1024;

    /**
     * 发布者的状态
     */
    private static final AtomicBoolean CLOSED = new AtomicBoolean(false);

    /**
     * 构造发布者的工厂
     */
    private static BiFunction<Class<? extends Event>, Integer, EventPublisher> publisherFactory = null;

    /**
     * 通知中心的实例
     */
    private static final NotifyCenter INSTANCE = new NotifyCenter();

    /**
     * 默认的多事件发布者
     */
    private DefaultSharePublisher sharePublisher;

    /**
     * 默认的单事件发布者类型
     * 此处并未直接指定单事件发布者是谁,只是限定了它的类别
     * 因为单事件发布者一个发布者只负责一个事件,因此会存在
     * 多个发布者实例,后面按需创建,并缓存在publisherMap
     */
    private static Class<? extends EventPublisher> clazz = null;

    /**
     * Publisher management container.
     * 单事件发布者存储容器
     */
    private final Map<String, EventPublisher> publisherMap = new ConcurrentHashMap<String, EventPublisher>(16);
	
	// 省略部分代码
}

可以看到它初始化了一个通知中心的实例,这里是单例模式。定义了发布者。订阅者是保存在发布者的内部,而发布者又保存在通知者的内部。这样就组成了一套完整的事件发布机制。

静态代码块

static {

	// 初始化DefaultPublisher的queue容量值
	// Internal ArrayBlockingQueue buffer size. For applications with high write throughput,
	// this value needs to be increased appropriately. default value is 16384
	String ringBufferSizeProperty = "nacos.core.notify.ring-buffer-size";
	ringBufferSize = Integer.getInteger(ringBufferSizeProperty, 16384);
	
	// 初始化DefaultSharePublisher的queue容量值
	// The size of the public publisher's message staging queue buffer
	String shareBufferSizeProperty = "nacos.core.notify.share-buffer-size";
	shareBufferSize = Integer.getInteger(shareBufferSizeProperty, 1024);
	
	// 使用Nacos SPI机制获取事件发布者
	final Collection<EventPublisher> publishers = NacosServiceLoader.load(EventPublisher.class);
	
	// 获取迭代器
	Iterator<EventPublisher> iterator = publishers.iterator();
	
	if (iterator.hasNext()) {
		clazz = iterator.next().getClass();
	} else {
		// 若为空,则使用默认的发布器(单事件发布者)
		clazz = DefaultPublisher.class;
	}
	
	// 声明发布者工厂为一个函数,用于创建发布者实例
	publisherFactory = new BiFunction<Class<? extends Event>, Integer, EventPublisher>() {
		
		/**
		 * 为指定类型的事件创建一个单事件发布者对象
		 * @param cls       事件类型
		 * @param buffer    发布者内部队列初始容量
		 * @return
		 */
		@Override
		public EventPublisher apply(Class<? extends Event> cls, Integer 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);
			}
		}
	};
	
	try {
		// 初始化多事件发布者
		// Create and init DefaultSharePublisher instance.
		INSTANCE.sharePublisher = new DefaultSharePublisher();
		INSTANCE.sharePublisher.init(SlowEvent.class, shareBufferSize);
		
	} catch (Throwable ex) {
		LOGGER.error("Service class newInstance has error : {}", ex);
	}

	// 增加关闭钩子,用于关闭Publisher
	ThreadUtils.addShutdownHook(new Runnable() {
		@Override
		public void run() {
			shutdown();
		}
	});

    }

在静态代码块中主要就做了两件事:

初始化单事件发布者:可以由用户扩展指定(通过Nacos SPI机制),也可以是Nacos默认的(DefaultPublisher)。

初始化多事件发布者:DefaultSharePublisher。

注册订阅者

注册订阅者实际上就是将Subscriber添加到Publisher中。因为事件的发布是靠发布者来通知它内部的所有订阅者。

/**
 * Register a Subscriber. If the Publisher concerned by the Subscriber does not exist, then PublihserMap will
 * preempt a placeholder Publisher first.
 *
 * @param consumer subscriber
 * @param <T>      event type
 */
public static <T> void registerSubscriber(final Subscriber consumer) {
	
	// 若想监听多个事件,实现SmartSubscriber.subscribeTypes()方法,在里面返回多个事件的列表即可
	// 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) {
		// 获取事件列表
		for (Class<? extends Event> subscribeType : ((SmartSubscriber) consumer).subscribeTypes()) {
			// 判断它的事件类型来决定采用哪种Publisher,多事件订阅者由多事件发布者调度
			// 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);
			}
		}
		return;
	}
	
	// 单事件的订阅者注册
	final Class<? extends Event> subscribeType = consumer.subscribeType();
	// 防止误使用,万一有人在使用单事件订阅者Subscriber的时候传入了SlowEvent则可以在此避免
	if (ClassUtils.isAssignableFrom(SlowEvent.class, subscribeType)) {
		INSTANCE.sharePublisher.addSubscriber(consumer, subscribeType);
		// 添加完毕返回
		return;
	}

	// 注册到单事件发布者中
	addSubscriber(consumer, subscribeType);
}

/**
 * 单事件发布者添加订阅者
 * Add a subscriber to publisher.
 * @param consumer      subscriber instance.
 * @param subscribeType subscribeType.
 */
private static void addSubscriber(final Subscriber consumer, Class<? extends Event> subscribeType) {
	// 获取类的规范名称,实际上就是包名加类名,作为topic
	final String topic = ClassUtils.getCanonicalName(subscribeType);
	synchronized (NotifyCenter.class) {
		// MapUtils.computeIfAbsent is a unsafe method.
		
		/**
		 * 生成指定类型的发布者,并将其放入publisherMap中
		 * 使用topic为key从publisherMap获取数据,若为空则使用publisherFactory函数并传递subscribeType和ringBufferSize来实例
		 * 化一个clazz类型的发布者对象,使用topic为key放入publisherMap中,实际上就是为每一个类型的事件创建一个发布者。具体
		 * 可查看publisherFactory的逻辑。
		 */
		MapUtil.computeIfAbsent(INSTANCE.publisherMap, topic, publisherFactory, subscribeType, ringBufferSize);
	}
	// 获取生成的发布者对象,将订阅者添加进去
	EventPublisher publisher = INSTANCE.publisherMap.get(topic);
	publisher.addSubscriber(consumer);
}

提示:
单事件发布者容器内的存储状态为: 事件类型的完整限定名 -> DefaultPublisher.
例如:
com.alibaba.nacos.core.cluster.MembersChangeEvent -> {DefaultPublisher@6839} “Thread[nacos.publisher-com.alibaba.nacos.core.cluster.MembersChangeEvent,5,main]”

注册发布者

实际上并没有直接的注册发布者这个概念,通过前面的章节你肯定知道发布者就两种类型:单事件发布者、多事件发布者。单事件发布者直接就一个实例,多事件发布者会根据事件类型创建不同的实例,存储于publisherMap中。它已经在通知中心了,因此并不需要有刻意的注册动作。需要使用的时候
直接取即可。

注册事件

注册事件实际上就是将具体的事件和具体的发布者进行关联,发布者有2种类型,那么事件也一定是两种类型了(事件的类型这里说的是分类,服务于单事件发布者的事件和服务于多事件发布者的事件)。

/**
 * Register publisher.
 *
 * @param eventType    class Instances type of the event type.
 * @param queueMaxSize the publisher's queue max size.
 */
public static EventPublisher registerToPublisher(final Class<? extends Event> eventType, final int queueMaxSize) {

	// 慢事件由多事件发布者处理
	if (ClassUtils.isAssignableFrom(SlowEvent.class, eventType)) {
		return INSTANCE.sharePublisher;
	}
	// 若不是慢事件,因为它可以存在多个不同的类型,因此需要判断对应的发布者是否存在
	final String topic = ClassUtils.getCanonicalName(eventType);
	synchronized (NotifyCenter.class) {
		// 当前传入的事件类型对应的发布者,有则忽略无则新建
		MapUtil.computeIfAbsent(INSTANCE.publisherMap, topic, publisherFactory, eventType, queueMaxSize);
	}
	return INSTANCE.publisherMap.get(topic);
}

这里并未有注册动作,若是SlowEvent则直接返回了,为何呢?这里再理一下关系,事件的实际用途是由订阅者来决定的,由订阅者来执行对应事件触发后的操作,事件和发布者并没有直接关系。而多事件发布者呢,它是一个发布者来处理所有的事件和订阅者(事件:订阅者,一对多的关系),这个事件都没人订阅何谈发布呢?因此单纯的注册事件并没有实际意义。反观一次只能处理一个事件的单事件处理器(DefaultPublisher)则需要一个事件对应一个发布者,即便这个事件没有人订阅,也可以缓存起来。

注销订阅者

注销的操作基本上就是注册的反向操作。

public static <T> void deregisterSubscriber(final Subscriber consumer) {
	// 若是多事件订阅者
	if (consumer instanceof SmartSubscriber) {
		// 获取事件列表
		for (Class<? extends Event> subscribeType : ((SmartSubscriber) consumer).subscribeTypes()) {
			// 若是慢事件
			if (ClassUtils.isAssignableFrom(SlowEvent.class, subscribeType)) {
				// 从多事件发布者中移除
				INSTANCE.sharePublisher.removeSubscriber(consumer, subscribeType);
			} else {
				// 从单事件发布者中移除
				removeSubscriber(consumer, subscribeType);
			}
		}
		return;
	}
	
	// 若是单事件订阅者
	final Class<? extends Event> subscribeType = consumer.subscribeType();
	// 判断是否是慢事件
	if (ClassUtils.isAssignableFrom(SlowEvent.class, subscribeType)) {
		INSTANCE.sharePublisher.removeSubscriber(consumer, subscribeType);
		return;
	}
	
	// 调用移除方法
	if (removeSubscriber(consumer, subscribeType)) {
		return;
	}
	throw new NoSuchElementException("The subscriber has no event publisher");
}

private static boolean removeSubscriber(final Subscriber consumer, Class<? extends Event> subscribeType) {
	// 获取topic
	final String topic = ClassUtils.getCanonicalName(subscribeType);
	// 根据topic获取对应的发布者
	EventPublisher eventPublisher = INSTANCE.publisherMap.get(topic);
	if (eventPublisher != null) {
		// 从发布者中移除订阅者
		eventPublisher.removeSubscriber(consumer);
		return true;
	}
	return false;
}

注销发布者

注销发布者主要针对于单事件发布者来说的,因为多事件发布者只有一个实例,它需要处理多个事件类型,因此发布者不能移除。而单事件发布者一个发布者对应一个事件类型,因此某个类型的事件不需要处理的时候则需要将对应的发布者移除。

public static void deregisterPublisher(final Class<? extends Event> eventType) {
	// 获取topic
	final String topic = ClassUtils.getCanonicalName(eventType);
	// 根据topic移除对应的发布者
	EventPublisher publisher = INSTANCE.publisherMap.remove(topic);
	try {
		// 调用关闭方法
		publisher.shutdown();
	} catch (Throwable ex) {
		LOGGER.error("There was an exception when publisher shutdown : {}", ex);
	}
}

public void shutdown() {
	// 标记关闭
	this.shutdown = true;
	// 清空缓存
	this.queue.clear();
}

发布事件

发布事件的本质就是不同类型的发布者来调用内部维护的订阅者的onEvent()方法。

private static boolean publishEvent(final Class<? extends Event> eventType, final Event event) {
	
	// 慢事件处理
	if (ClassUtils.isAssignableFrom(SlowEvent.class, eventType)) {
		return INSTANCE.sharePublisher.publish(event);
	}
	
	// 常规事件处理
	final String topic = ClassUtils.getCanonicalName(eventType);
	
	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;
}

总结

在Nacos中的事件发布分为两条线:单一事件处理、多事件处理。围绕这两条线又有负责单一类型事件的订阅者、发布者,也有负责多事件的订阅者、发布者。区分开来两种类型便很容易理解。

Nacos源码分析-事件发布机制

上图展示了在通知中心中不同类型的事件、订阅者、发布者的存储状态。

多事件发布者:

  • 发布者和事件的关系是一对多
  • 事件和订阅者的关系是一对多
  • 发布者和订阅者的关系是一对多
  • 事件类型为SlowEvent, 订阅者类型是SmartSubscriber

单事件发布者

  • 发布者和事件的关系是一对一
  • 事件和订阅者的关系是一对多
  • 发布者和订阅者的关系是一对多
  • 事件类型为Event,订阅者类型是Subscriber

程序员灯塔
转载请注明原文链接:Nacos源码分析-事件发布机制
喜欢 (0)