共计 9962 个字符,预计需要花费 25 分钟才能阅读完成。
学习不必那么功利,二师兄带你从更高维度轻松浏览源码~
上篇文章,咱们剖析了 Nacos 客户端订阅的外围流程:Nacos 客户端通过一个定时工作,每 6 秒从注册核心获取实例列表,当发现实例发生变化时,公布变更事件,订阅者进行业务解决,而后更新内存中和本地的缓存中的实例。
这篇文章为服务订阅的第二篇,咱们重点来剖析,定时工作获取到最新实例列表之后,整个事件机制是如何解决的。
回顾整个流程
先回顾一下客户端服务订阅的根本流程:
在第一步调用 subscribe 办法时,会订阅一个 EventListener 事件。而在定时工作 UpdateTask 定时获取实例列表之后,会调用 ServiceInfoHolder#processServiceInfo 办法对 ServiceInfo 进行本地解决,这其中就包含和事件处理。
监听事件的注册
在 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 便是进行具体的事件注册逻辑。追进去看一下实现源码:
// InstancesChangeNotifier
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 属性当中了。
这里的数据结构为 Map,key 为服务实例信息的拼接,value 为监听事件的汇合。
事件注册流程就这么简略。这里有一个双重查看锁的实际案例,不晓得你留意到没?能够学习一下。
ServiceInfo 的解决
下面实现了事件的注册,当初就追溯一下触发事件的起源。UpdateTask 中获取到最新实例会进行本地化解决,局部代码如下:
// 获取缓存的 service 信息
ServiceInfo serviceObj = serviceInfoHolder.getServiceInfoMap().get(serviceKey);
if (serviceObj == null) {
// 依据 serviceName 从注册核心服务端获取 Service 信息
serviceObj = namingClientProxy.queryInstancesOfService(serviceName, groupName, clusters, 0, false);
serviceInfoHolder.processServiceInfo(serviceObj);
lastRefTime = serviceObj.getLastRefTime();
return;
}
这部分逻辑在上篇文章中曾经剖析过了,这里重点看 serviceInfoHolder#processServiceInfo 中的业务逻辑解决。先看流程图,而后看代码。
上述逻辑简略说就是:判断一下新的 ServiceInfo 数据是否正确,是否产生了变动。如果数据格式正确,且产生的变动,那就公布一个 InstancesChangeEvent 事件,同时将 ServiceInfo 写入本地缓存。
上面看一下代码实现:
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));
}
// 通过 prometheus-simpleclient 监控服务缓存 Map 的大小
MetricsMonitor.getServiceInfoMapSizeMonitor().set(serviceInfoMap.size());
// 服务实例已变更
if (changed) {NAMING_LOGGER.info("current ips:(" + serviceInfo.ipCount() + ") service:" + 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 中的外围流程如下:
NotifyCenter 中进行事件公布,公布的外围逻辑是:
- 依据 InstancesChangeEvent 事件类型,取得对应的 CanonicalName;
- 将 CanonicalName 作为 Key,从 NotifyCenter#publisherMap 中获取对应的事件发布者(EventPublisher);
- EventPublisher 将 InstancesChangeEvent 事件进行公布。
NotifyCenter 中的外围代码实现如下:
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) {
// EventPublisher 将 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 来进行构建。
public static EventPublisher registerToPublisher(final Class<? extends Event> eventType, final int queueMaxSize) {return registerToPublisher(eventType, DEFAULT_PUBLISHER_FACTORY, queueMaxSize);
}
如果查看 NotifyCenter 中动态代码块,会发现 DEFAULT_PUBLISHER_FACTORY 默认构建的 EventPublisher 为 DefaultPublisher。
至此,咱们得悉,在 NotifyCenter 中它保护了事件名称和事件发布者的关系,而默认的事件发布者为 DefaultPublisher。
DefaultPublisher 的事件公布
查看 DefaultPublisher 的源码,会发现它继承自 Thread,也就是说它是一个线程类。同时,它又实现了 EventPublisher,也就是咱们后面提到的发布者。
public class DefaultPublisher extends Thread implements EventPublisher {}
在 DefaultPublisher 的 init 办法实现如下:
@Override
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<>(bufferSize);
start();}
也就是说,当 DefaultPublisher 被初始化时,是以守护线程的形式运作的,其中还初始化了一个阻塞队列,队列的默认大小为 16384。
最初调用了 start 办法:
@Override
public synchronized void start() {if (!initialized) {
// start just called once
super.start();
if (queueMaxSize == -1) {queueMaxSize = ringBufferSize;}
initialized = true;
}
}
start 办法中调用了 super.start,此时等于启动了线程,会执行对应的 run 办法。
run 办法中只调用了如下办法:
void openEventHandler() {
try {
// This variable is defined to resolve the problem which message overstock in the queue.
int waitTimes = 60;
// for 死循环不断的从队列中取出 Event,并告诉订阅者 Subscriber 执行 Event
// To ensure that messages are not lost, enable EventHandler when
// waiting for the first Subscriber to register
for (; ;) {if (shutdown || hasSubscriber() || waitTimes <= 0) {break;}
ThreadUtils.sleep(1000L);
waitTimes--;
}
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 办法实现:
@Override
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;
}
能够看到,DefaultPublisher 的 publish 办法确实就是往阻塞队列中存入事件。这里有个分支逻辑,如果存入失败,会间接调用 receiveEvent,和从队列中取出事件执行的办法一样。能够了解为,如果向队列中存入失败,则立刻执行,不走队列了。
最初,再来看看 receiveEvent 办法的实现:
void receiveEvent(Event event) {final long currentEventSequence = event.sequence();
if (!hasSubscriber()) {LOGGER.warn("[NotifyCenter] the {} is lost, because there is no subscriber.");
return;
}
// 告诉订阅者执行 Event
// Notification single event listener
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);
}
}
这里最次要的逻辑就是遍历 DefaultPublisher 的 subscribers(订阅者汇合),而后执行告诉订阅者的办法。
那么有敌人要问了这 subscribers 中的订阅者哪里来的呢?这个还要回到 NacosNamingService 的 init 办法中:
// 将 Subscribe 注册到 Publisher
NotifyCenter.registerSubscriber(changeNotifier);
该办法最终会调用 NotifyCenter 的 addSubscriber 办法:
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);
}
}
其中外围逻辑就是将订阅事件、发布者、订阅者三者进行绑定。而发布者与事件通过 Map 进行保护、发布者与订阅者通过关联关系进行保护。
发布者找到了,事件也有了,最初看一下 notifySubscriber 办法:
@Override
public void notifySubscriber(final Subscriber subscriber, final Event event) {LOGGER.debug("[NotifyCenter] the {} will received by {}", event, subscriber);
// 执行订阅者 Event
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);
}
}
}
逻辑比较简单,如果订阅者定义了 Executor,那么应用它定义的 Executor 进行事件的执行,如果没有,那就创立一个线程进行执行。
至此,整个服务订阅的事件机制实现。
小结
整体来看,整个服务订阅的事件机制还是比较复杂的,因为用到了事件的模式,逻辑就比拟绕,而且这期间还掺杂了守护线程,死循环,阻塞队列等。须要重点了解 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 事件便是执行订阅时传入的事件;
对于 Nacos Client 服务定义的事件机制就将这么多,下篇咱们来讲讲故障转移和缓存的实现。
博主简介:《SpringBoot 技术底细》技术图书作者,热爱钻研技术,写技术干货文章。
公众号:「程序新视界」,博主的公众号,欢送关注~
技术交换:请分割博主微信号:zhuan2quan