NacosNamingService中subscribe及unsubscribe的原理和使用方法
这篇文章主要讲解了“NacosNamingService中subscribe及unsubscribe的原理和使用方法”,文中的讲解内容简单清晰,易于学习与理解,下面请大家跟着小编的思路慢慢深入,一起来研究和学习“NacosNamingService中subscribe及unsubscribe的原理和使用方法”吧!
“只有客户发展了,才有我们的生存与发展!”这是成都创新互联公司的服务宗旨!把网站当作互联网产品,产品思维更注重全局思维、需求分析和迭代思维,在网站建设中就是为了建设一个不仅审美在线,而且实用性极高的网站。创新互联对成都做网站、成都网站建设、网站制作、网站开发、网页设计、网站优化、网络推广、探索永无止境。
序
本文主要研究一下NacosNamingService的subscribe及unsubscribe
NacosNamingService
nacos-1.1.3/client/src/main/java/com/alibaba/nacos/client/naming/NacosNamingService.java
public class NacosNamingService implements NamingService { private static final String DEFAULT_PORT = "8080"; private static final long DEFAULT_HEART_BEAT_INTERVAL = TimeUnit.SECONDS.toMillis(5); /** * Each Naming instance should have different namespace. */ private String namespace; private String endpoint; private String serverList; private String cacheDir; private String logName; private HostReactor hostReactor; private BeatReactor beatReactor; private EventDispatcher eventDispatcher; private NamingProxy serverProxy; //...... @Override public void subscribe(String serviceName, EventListener listener) throws NacosException { subscribe(serviceName, new ArrayList(), listener); } @Override public void subscribe(String serviceName, String groupName, EventListener listener) throws NacosException { subscribe(serviceName, groupName, new ArrayList (), listener); } @Override public void subscribe(String serviceName, List clusters, EventListener listener) throws NacosException { subscribe(serviceName, Constants.DEFAULT_GROUP, clusters, listener); } @Override public void subscribe(String serviceName, String groupName, List clusters, EventListener listener) throws NacosException { eventDispatcher.addListener(hostReactor.getServiceInfo(NamingUtils.getGroupedName(serviceName, groupName), StringUtils.join(clusters, ",")), StringUtils.join(clusters, ","), listener); } @Override public void unsubscribe(String serviceName, EventListener listener) throws NacosException { unsubscribe(serviceName, new ArrayList (), listener); } @Override public void unsubscribe(String serviceName, String groupName, EventListener listener) throws NacosException { unsubscribe(serviceName, groupName, new ArrayList (), listener); } @Override public void unsubscribe(String serviceName, List clusters, EventListener listener) throws NacosException { unsubscribe(serviceName, Constants.DEFAULT_GROUP, clusters, listener); } @Override public void unsubscribe(String serviceName, String groupName, List clusters, EventListener listener) throws NacosException { eventDispatcher.removeListener(NamingUtils.getGroupedName(serviceName, groupName), StringUtils.join(clusters, ","), listener); } //...... }
subscribe方法执行eventDispatcher.addListener;unsubscribe方法执行eventDispatcher.removeListener
EventDispatcher
nacos-1.1.3/client/src/main/java/com/alibaba/nacos/client/naming/core/EventDispatcher.java
public class EventDispatcher { private ExecutorService executor = null; private BlockingQueuechangedServices = new LinkedBlockingQueue (); private ConcurrentMap > observerMap = new ConcurrentHashMap >(); public EventDispatcher() { 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; } }); executor.execute(new Notifier()); } public void addListener(ServiceInfo serviceInfo, String clusters, EventListener listener) { NAMING_LOGGER.info("[LISTENER] adding " + serviceInfo.getName() + " with " + clusters + " to listener map"); List observers = Collections.synchronizedList(new ArrayList ()); observers.add(listener); observers = observerMap.putIfAbsent(ServiceInfo.getKey(serviceInfo.getName(), clusters), observers); if (observers != null) { observers.add(listener); } serviceChanged(serviceInfo); } public void removeListener(String serviceName, String clusters, EventListener listener) { NAMING_LOGGER.info("[LISTENER] removing " + serviceName + " with " + clusters + " from listener map"); List observers = observerMap.get(ServiceInfo.getKey(serviceName, clusters)); if (observers != null) { Iterator iter = observers.iterator(); while (iter.hasNext()) { EventListener oldListener = iter.next(); if (oldListener.equals(listener)) { iter.remove(); } } if (observers.isEmpty()) { observerMap.remove(ServiceInfo.getKey(serviceName, clusters)); } } } public List getSubscribeServices() { List serviceInfos = new ArrayList (); for (String key : observerMap.keySet()) { serviceInfos.add(ServiceInfo.fromKey(key)); } return serviceInfos; } public void serviceChanged(ServiceInfo serviceInfo) { if (serviceInfo == null) { return; } changedServices.add(serviceInfo); } private class Notifier implements Runnable { @Override public void run() { while (true) { ServiceInfo serviceInfo = null; try { serviceInfo = changedServices.poll(5, TimeUnit.MINUTES); } catch (Exception ignore) { } if (serviceInfo == null) { continue; } try { List listeners = observerMap.get(serviceInfo.getKey()); if (!CollectionUtils.isEmpty(listeners)) { for (EventListener listener : listeners) { List 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); } } } } public void setExecutor(ExecutorService executor) { ExecutorService oldExecutor = this.executor; this.executor = executor; oldExecutor.shutdown(); } }
EventDispatcher的构造器创建了executor,并执行Notifier;Notifier使用一个while true循环不断执行changedServices.poll(5, TimeUnit.MINUTES)拉取serviceInfo,拉取到的话会从observerMap取出对应的EventListener列表,然后挨个回调listener.onEvent方法
addListener方法则是往observerMap创建或添加observers,然后执行serviceChanged方法;removeListener则是从observerMap移除指定的listener,如果指定key的listener列表为空则删除该key
serviceChanged方法会往changedServices添加serviceInfo;之后Notifier异步线程可用拉取信息执行listener.onEvent回调
小结
NacosNamingService的subscribe方法执行eventDispatcher.addListener;unsubscribe方法执行eventDispatcher.removeListener
感谢各位的阅读,以上就是“NacosNamingService中subscribe及unsubscribe的原理和使用方法”的内容了,经过本文的学习后,相信大家对NacosNamingService中subscribe及unsubscribe的原理和使用方法这一问题有了更深刻的体会,具体使用情况还需要大家实践验证。这里是创新互联,小编将为大家推送更多相关知识点的文章,欢迎关注!
标题名称:NacosNamingService中subscribe及unsubscribe的原理和使用方法
网站URL:http://scyanting.com/article/gopihi.html