NacosClient服务订阅机制之核心流程

本文转载自微信公众号「程序新视界」，作者二师兄。转载本文请联系程序新视界公众号。

说起Nacos的服务订阅机制，对此不了解的朋友，可能感觉非常神秘，这篇文章就大家深入浅出的了解一下Nacos 2.0客户端的订阅实现。由于涉及到的内容比较多，就分几篇来讲，本篇为第一篇。

Nacos订阅概述

Nacos的订阅机制，如果用一句话来描述就是：Nacos客户端通过一个定时任务，每6秒从注册中心获取实例列表，当发现实例发生变化时，发布变更事件，订阅者进行业务处理。该更新实例的更新实例，该更新本地缓存的更新本地缓存。

nacos

上图画出了订阅方法的主线流程，涉及的内容较多，处理细节复杂。这里只用把握住核心部分即可。下面就通过代码和流程图来逐步分析上述过程。

从订阅到定时任务开启

我们这里聊的订阅机制，其实本质上就是服务发现的准实时感知。上面已经看到了当执行订阅方法时，会触发定时任务，定时去拉服务器端的数据。所以，本质上，订阅机制就是实现服务发现的一种方式，对照的方式就是直接查询接口了。

NacosNamingService中暴露的许多重载的subscribe，重载的目的就是让大家少写一些参数，这些参数呢，Nacos给默认处理了。最终这些重载方法都会调用到下面这个方法：

 
 
 
 
    
  
  
  
  
  1. // NacosNamingService 
    2. 
  
  
  
  
  2. public void subscribe(String serviceName, String groupName, List clusters, EventListener listener) 
    3. 
  
  
  
  
  3.         throws NacosException { 
    4. 
  
  
  
  
  4.     if (null == listener) { 
    5. 
  
  
  
  
  5.         return; 
    6. 
  
  
  
  
  6.     } 
    7. 
  
  
  
  
  7.     String clusterString = StringUtils.join(clusters, ","); 
    8. 
  
  
  
  
  8.     changeNotifier.registerListener(groupName, serviceName, clusterString, listener); 
    9. 
  
  
  
  
  9.     clientProxy.subscribe(serviceName, groupName, clusterString); 
    10. 
  
  
  
  
  10. } 
    11.

方法中的事件监听我们暂时不聊，直接看subscribe方法，这里clientProxy类型为NamingClientProxyDelegate。实例化NacosNamingService时该类被实例化，前面章节中已经讲到，不再赘述。

而clientProxy.subscribe方法在NamingClientProxyDelegate中实现：

 
 
 
 
    
  
  
  
  
  1. // NamingClientProxyDelegate 
    2. 
  
  
  
  
  2. @Override 
    3. 
  
  
  
  
  3. public ServiceInfo subscribe(String serviceName, String groupName, String clusters) throws NacosException { 
    4. 
  
  
  
  
  4.     String serviceNameWithGroup = NamingUtils.getGroupedName(serviceName, groupName); 
    5. 
  
  
  
  
  5.     String serviceKey = ServiceInfo.getKey(serviceNameWithGroup, clusters); 
    6. 
  
  
  
  
  6.     // 获取缓存中的ServiceInfo 
    7. 
  
  
  
  
  7.     ServiceInfo result = serviceInfoHolder.getServiceInfoMap().get(serviceKey); 
    8. 
  
  
  
  
  8.     if (null == result) { 
    9. 
  
  
  
  
  9.         // 如果为null，则进行订阅逻辑处理，基于gRPC协议 
    10. 
  
  
  
  
  10.         result = grpcClientProxy.subscribe(serviceName, groupName, clusters); 
    11. 
  
  
  
  
  11.     } 
    12. 
  
  
  
  
  12.     // 定时调度UpdateTask 
    13. 
  
  
  
  
  13.     serviceInfoUpdateService.scheduleUpdateIfAbsent(serviceName, groupName, clusters); 
    14. 
  
  
  
  
  14.     // ServiceInfo本地缓存处理 
    15. 
  
  
  
  
  15.     serviceInfoHolder.processServiceInfo(result); 
    16. 
  
  
  
  
  16.     return result; 
    17. 
  
  
  
  
  17. } 
    18.

这段方法是不是眼熟啊?对的，在前面分析《Nacos Client服务发现》时我们已经讲过了。看来殊途同归，查询服务列表和订阅最终都调用了同一个方法。

上篇讲了其他流程，我们这里重点看任务调度：

 
 
 
 
    
  
  
  
  
  1. // ServiceInfoUpdateService 
    2. 
  
  
  
  
  2. public void scheduleUpdateIfAbsent(String serviceName, String groupName, String clusters) { 
    3. 
  
  
  
  
  3.     String serviceKey = ServiceInfo.getKey(NamingUtils.getGroupedName(serviceName, groupName), clusters); 
    4. 
  
  
  
  
  4.     if (futureMap.get(serviceKey) != null) { 
    5. 
  
  
  
  
  5.         return; 
    6. 
  
  
  
  
  6.     } 
    7. 
  
  
  
  
  7.     synchronized (futureMap) { 
    8. 
  
  
  
  
  8.         if (futureMap.get(serviceKey) != null) { 
    9. 
  
  
  
  
  9.             return; 
    10. 
  
  
  
  
  10.         } 
    11. 
  
  
  
  
  11.         // 构建UpdateTask 
    12. 
  
  
  
  
  12.         ScheduledFuture future = addTask(new UpdateTask(serviceName, groupName, clusters)); 
    13. 
  
  
  
  
  13.         futureMap.put(serviceKey, future); 
    14. 
  
  
  
  
  14.     } 
    15. 
  
  
  
  
  15. } 
    16.

该方法包含了构建serviceKey、通过serviceKey判重，最后添加UpdateTask。

而其中的addTask的实现就是发起了一个定时任务：

 
 
 
 
    
  
  
  
  
  1. // ServiceInfoUpdateService 
    2. 
  
  
  
  
  2. private synchronized ScheduledFuture addTask(UpdateTask task) { 
    3. 
  
  
  
  
  3.     return executor.schedule(task, DEFAULT_DELAY, TimeUnit.MILLISECONDS); 
    4. 
  
  
  
  
  4. } 
    5.

定时任务延时1秒执行。

跟踪到这里就告一阶段了。核心功能只有两个：调用订阅方法和发起定时任务。

定时任务都干了啥

UpdateTask封装了订阅机制的核心业务逻辑，先来通过一张流程图看一下都做了啥。

nacos

有了上述流程图，基本就很清晰的了解UpdateTask所做的事情了。直接贴出run方法的所有代码：

 
 
 
 
    
  
  
  
  
  1. public void run() { 
    2. 
  
  
  
  
  2.     long delayTime = DEFAULT_DELAY; 
    3. 
  
  
  
  
  3.  
    4. 
  
  
  
  
  4.     try { 
    5. 
  
  
  
  
  5.         // 判断该注册的Service是否被订阅，如果没有订阅则不再执行 
    6. 
  
  
  
  
  6.         if (!changeNotifier.isSubscribed(groupName, serviceName, clusters) && !futureMap.containsKey(serviceKey)) { 
    7. 
  
  
  
  
  7.             NAMING_LOGGER 
    8. 
  
  
  
  
  8.                     .info("update task is stopped, service:" + groupedServiceName + ", clusters:" + clusters); 
    9. 
  
  
  
  
  9.             return; 
    10. 
  
  
  
  
  10.         } 
    11. 
  
  
  
  
  11.  
    12. 
  
  
  
  
  12.         // 获取缓存的service信息 
    13. 
  
  
  
  
  13.         ServiceInfo serviceObj = serviceInfoHolder.getServiceInfoMap().get(serviceKey); 
    14. 
  
  
  
  
  14.         if (serviceObj == null) { 
    15. 
  
  
  
  
  15.             // 根据serviceName从注册中心服务端获取Service信息 
    16. 
  
  
  
  
  16.             serviceObj = namingClientProxy.queryInstancesOfService(serviceName, groupName, clusters, 0, false); 
    17. 
  
  
  
  
  17.             serviceInfoHolder.processServiceInfo(serviceObj); 
    18. 
  
  
  
  
  18.             lastRefTime = serviceObj.getLastRefTime(); 
    19. 
  
  
  
  
  19.             return; 
    20. 
  
  
  
  
  20.         } 
    21. 
  
  
  
  
  21.  
    22. 
  
  
  
  
  22.         // 过期服务（服务的最新更新时间小于等于缓存刷新时间），从注册中心重新查询 
    23. 
  
  
  
  
  23.         if (serviceObj.getLastRefTime() <= lastRefTime) { 
    24. 
  
  
  
  
  24.             serviceObj = namingClientProxy.queryInstancesOfService(serviceName, groupName, clusters, 0, false); 
    25. 
  
  
  
  
  25.             // 处理Service消息 
    26. 
  
  
  
  
  26.             serviceInfoHolder.processServiceInfo(serviceObj); 
    27. 
  
  
  
  
  27.         } 
    28. 
  
  
  
  
  28.         // 刷新更新时间 
    29. 
  
  
  
  
  29.         lastRefTime = serviceObj.getLastRefTime(); 
    30. 
  
  
  
  
  30.         if (CollectionUtils.isEmpty(serviceObj.getHosts())) { 
    31. 
  
  
  
  
  31.             incFailCount(); 
    32. 
  
  
  
  
  32.             return; 
    33. 
  
  
  
  
  33.         } 
    34. 
  
  
  
  
  34.         // 下次更新缓存时间设置，默认为6秒 
    35. 
  
  
  
  
  35.         // TODO multiple time can be configured. 
    36. 
  
  
  
  
  36.         delayTime = serviceObj.getCacheMillis() * DEFAULT_UPDATE_CACHE_TIME_MULTIPLE; 
    37. 
  
  
  
  
  37.         // 重置失败数量为0 
    38. 
  
  
  
  
  38.         resetFailCount(); 
    39. 
  
  
  
  
  39.     } catch (Throwable e) { 
    40. 
  
  
  
  
  40.         incFailCount(); 
    41. 
  
  
  
  
  41.         NAMING_LOGGER.warn("[NA] failed to update serviceName: " + groupedServiceName, e); 
    42. 
  
  
  
  
  42.     } finally { 
    43. 
  
  
  
  
  43.         // 下次调度刷新时间，下次执行的时间与failCount有关 
    44. 
  
  
  
  
  44.         // failCount=0，则下次调度时间为6秒，最长为1分钟 
    45. 
  
  
  
  
  45.         // 即当无异常情况下缓存实例的刷新时间是6秒 
    46. 
  
  
  
  
  46.         executor.schedule(this, Math.min(delayTime << failCount, DEFAULT_DELAY * 60), TimeUnit.MILLISECONDS); 
    47. 
  
  
  
  
  47.     } 
    48. 
  
  
  
  
  48. } 
    49.

首先在判断服务是否是被订阅过，实现方法是ChangeNotifier#isSubscribed：

 
 
 
 
    
  
  
  
  
  1. public boolean isSubscribed(String groupName, String serviceName, String clusters) { 
    2. 
  
  
  
  
  2.     String key = ServiceInfo.getKey(NamingUtils.getGroupedName(serviceName, groupName), clusters); 
    3. 
  
  
  
  
  3.     ConcurrentHashSet eventListeners = listenerMap.get(key); 
    4. 
  
  
  
  
  4.     return CollectionUtils.isNotEmpty(eventListeners); 
    5. 
  
  
  
  
  5. } 
    6.

查看该方法的源码会发现，这里的listenerMap正是最开始的subscribe方法中registerListener注册的EventListener。

run方法后面的业务处理基本上都雷同了，先判断缓存是否有ServiceInfo信息，如果没有则查询注册中心、处理ServiceInfo、更新上次处理时间。

而下面判断ServiceInfo是否失效，正是通过“上次更新时间”与当前ServiceInfo中的“上次更新时间”做比较来判断。如果失效，也会查询注册中心、处理ServiceInfo、更新上次处理时间等一系列操作。

业务逻辑最后会计算下一次定时任务的执行时间，通过delayTime来延迟执行。delayTime默认为 1000L * 6，也就是6秒。而在finally里面真的发起下一次定时任务。当出现异常时，下次执行的时间与失败次数有关，但最长不超过1分钟。

小结

这一篇我们讲了Nacos客户端服务订阅机制的源码，主要有以下步骤：

第一步：订阅方法的调用，并进行EventListener的注册，后面UpdateTask要用来进行判断;

第二步：通过委托代理类来处理订阅逻辑，此处与获取实例列表方法使用了同一个方法;

第三步：通过定时任务执行UpdateTask方法，默认执行间隔为6秒，当发生异常时会延长，但不超过1分钟;

第四步：UpdateTask方法中会比较本地是否存在缓存，缓存是否过期。当不存在或过期时，查询注册中心，获取最新实例，更新最后获取时间，处理ServiceInfo。

第五步：重新计算定时任务时间，循环执行上述流程。

网站栏目：NacosClient服务订阅机制之核心流程
URL分享：http://www.shufengxianlan.com/qtweb/news30/379530.html

网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 创新互联

猜你还喜欢下面的内容

App开发知识

同城分类信息

Android开发　　 　自贡网站建设　　 　成都微信小程序开发　　 　自贡电信机房　　 　h5页面制作　　 　乐山服务器托管　　 　食品包装设计　　 　手机微信网站　　 　400电话申请　　 　正泰动物　　 　传美奇电器　　 　美美搬家　　 　四川和正源　　 　成都网站建设报价　　 　关键词优化　　 　眉山托管服务器　　 　内江做网站　　 　汕尾凯敏网站　　 　重庆机房托管　　 　成都网站建设