13518219792
这篇文章主要讲解了“NacosNamingService中subscribe及unsubscribe的原理和使用方法”，文中的讲解内容简单清晰，易于学习与理解，下面请大家跟着小编的思路慢慢深入，一起来研究和学习“NacosNamingService中subscribe及unsubscribe的原理和使用方法”吧！
“只有客户发展了，才有我们的生存与发展！”这是成都创新互联公司的服务宗旨！把网站当作互联网产品，产品思维更注重全局思维、需求分析和迭代思维，在网站建设中就是为了建设一个不仅审美在线，而且实用性极高的网站。创新互联对成都做网站、成都网站建设、网站制作、网站开发、网页设计、网站优化、网络推广、探索永无止境。
本文主要研究一下NacosNamingService的subscribe及unsubscribe
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
nacos-1.1.3/client/src/main/java/com/alibaba/nacos/client/naming/core/EventDispatcher.java
public class EventDispatcher {
private ExecutorService executor = null;
private BlockingQueue changedServices = 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的原理和使用方法这一问题有了更深刻的体会，具体使用情况还需要大家实践验证。这里是创新互联，小编将为大家推送更多相关知识点的文章，欢迎关注！
