咱们在 Nacos – NacosNamingService 初始化提过,NacosNamingService 对象创立的时候,会创立一个 EventDispatcher 对象。EventDispatcher 的构造方法如下,创立一个线程池,而后放入 Notifier 工作。
public EventDispatcher() {this.executor = Executors.newSingleThreadExecutor(new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {Thread thread = new Thread(r, "com.alibaba.nacos.naming.client.listener");
thread.setDaemon(true);
return thread;
}
});
this.executor.execute(new Notifier());
}
Notifier 的 Runnable 的类,所以放入线程池的时候,会执行 run 办法。他次要是从阻塞队列 changedServices 取出 ServiceInfo,而后依据 ServiceInfo 的 key 取出他对应的 EventListener 汇合,再执行 EventListener 的 onEvent 办法。
@Override
public void run() {while (!closed) {
ServiceInfo serviceInfo = null;
try {
// changedServices 是 LinkedBlockingQueue,从阻塞队列取值
serviceInfo = changedServices.poll(5, TimeUnit.MINUTES);
} catch (Exception ignore) { }
// 没取值,从新从阻塞队列取
if (serviceInfo == null) {continue;}
try {
// 从 observerMap 取到 EventListener 汇合
List<EventListener> listeners = observerMap.get(serviceInfo.getKey());
if (!CollectionUtils.isEmpty(listeners)) {for (EventListener listener : listeners) {
// 执行 onEvent 办法
List<Instance> hosts = Collections.unmodifiableList(serviceInfo.getHosts());
listener.onEvent(new NamingEvent(serviceInfo.getName(), serviceInfo.getGroupName(),
serviceInfo.getClusters(), hosts));
}
}
} catch (Exception e) {NAMING_LOGGER.error("[NA] notify error for service:" + serviceInfo.getName() + ", clusters:"
+ serviceInfo.getClusters(), e);
}
}
}
在 Nacos – 启动中,提到 NacosWatch 实例化的时候,就会调用 namingService.subscribe,他会调用 EventDispatcher#addListener 办法,在这里会把监听放入 observerMap 的 map 里,而后调用 serviceChanged 办法。
public void addListener(ServiceInfo serviceInfo, String clusters, EventListener listener) {NAMING_LOGGER.info("[LISTENER] adding" + serviceInfo.getName() + "with" + clusters + "to listener map");
List<EventListener> observers = Collections.synchronizedList(new ArrayList<EventListener>());
observers.add(listener);
observers = observerMap.putIfAbsent(ServiceInfo.getKey(serviceInfo.getName(), clusters), observers);
if (observers != null) {observers.add(listener);
}
serviceChanged(serviceInfo);
}
当 serviceChanged 被调用的时候,就会往阻塞队列存入 ServiceInfo。下面曾经晓得了有个循环工作始终从阻塞队列 changedServices 取值,这个值就是这么来的。
public void serviceChanged(ServiceInfo serviceInfo) {if (serviceInfo == null) {return;}
changedServices.add(serviceInfo);
}
总结
这个类有两个比拟重要的成员,一个是 observerMap,他的 key 是 serviceInfo.getKey(),value 是 EventListener 汇合。一个是 changedServices,寄存 serviceInfo 的 LinkedBlockingQueue 阻塞队列。这两个成员的关联关系通过 serviceInfo.getKey() 维持。
当调用 addListener 的时候,就会把 serviceInfo 存入到 changedServices,以及 serviceInfo.getKey() 和 EventListener 汇合存入到 observerMap。
while(true) 中,因为阻塞队列 changedServices 有值,就会从中拿到 serviceInfo,再通过 serviceInfo.getKey() 拿到 observerMap 对应的 EventListener 汇合,而后执行 EventListener 汇合的 EventListener.onEvent 办法。