av激情亚洲男人的天堂国语,日韩欧美精品一中文字幕,无码av一区二区三区无码,国产又色又爽又刺激的a片,国产又色又爽又刺激的a片

NacosClient服務(wù)訂閱機(jī)制之核心流程

本文轉(zhuǎn)載自微信公眾號「程序新視界」，作者二師兄。轉(zhuǎn)載本文請聯(lián)系程序新視界公眾號。

說起Nacos的服務(wù)訂閱機(jī)制，對此不了解的朋友，可能感覺非常神秘，這篇文章就大家深入淺出的了解一下Nacos 2.0客戶端的訂閱實現(xiàn)。由于涉及到的內(nèi)容比較多，就分幾篇來講，本篇為第一篇。

Nacos訂閱概述

Nacos的訂閱機(jī)制，如果用一句話來描述就是：Nacos客戶端通過一個定時任務(wù)，每6秒從注冊中心獲取實例列表，當(dāng)發(fā)現(xiàn)實例發(fā)生變化時，發(fā)布變更事件，訂閱者進(jìn)行業(yè)務(wù)處理。該更新實例的更新實例，該更新本地緩存的更新本地緩存。

nacos

上圖畫出了訂閱方法的主線流程，涉及的內(nèi)容較多，處理細(xì)節(jié)復(fù)雜。這里只用把握住核心部分即可。下面就通過代碼和流程圖來逐步分析上述過程。

從訂閱到定時任務(wù)開啟

我們這里聊的訂閱機(jī)制，其實本質(zhì)上就是服務(wù)發(fā)現(xiàn)的準(zhǔn)實時感知。上面已經(jīng)看到了當(dāng)執(zhí)行訂閱方法時，會觸發(fā)定時任務(wù)，定時去拉服務(wù)器端的數(shù)據(jù)。所以，本質(zhì)上，訂閱機(jī)制就是實現(xiàn)服務(wù)發(fā)現(xiàn)的一種方式，對照的方式就是直接查詢接口了。

NacosNamingService中暴露的許多重載的subscribe，重載的目的就是讓大家少寫一些參數(shù)，這些參數(shù)呢，Nacos給默認(rèn)處理了。最終這些重載方法都會調(diào)用到下面這個方法：

 
 
 
 
    
  
  
  
  
  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.

方法中的事件監(jiān)聽我們暫時不聊，直接看subscribe方法，這里clientProxy類型為NamingClientProxyDelegate。實例化NacosNamingService時該類被實例化，前面章節(jié)中已經(jīng)講到，不再贅述。

而clientProxy.subscribe方法在NamingClientProxyDelegate中實現(xiàn)：

 
 
 
 
    
  
  
  
  
  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，則進(jìn)行訂閱邏輯處理，基于gRPC協(xié)議 
    10. 
  
  
  
  
  10.         result = grpcClientProxy.subscribe(serviceName, groupName, clusters); 
    11. 
  
  
  
  
  11.     } 
    12. 
  
  
  
  
  12.     // 定時調(diào)度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服務(wù)發(fā)現(xiàn)》時我們已經(jīng)講過了?？磥硎馔就瑲w，查詢服務(wù)列表和訂閱最終都調(diào)用了同一個方法。

上篇講了其他流程，我們這里重點(diǎn)看任務(wù)調(diào)度：

 
 
 
 
    
  
  
  
  
  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.         // 構(gòu)建UpdateTask 
    12. 
  
  
  
  
  12.         ScheduledFuture future = addTask(new UpdateTask(serviceName, groupName, clusters)); 
    13. 
  
  
  
  
  13.         futureMap.put(serviceKey, future); 
    14. 
  
  
  
  
  14.     } 
    15. 
  
  
  
  
  15. } 
    16.

該方法包含了構(gòu)建serviceKey、通過serviceKey判重，最后添加UpdateTask。

而其中的addTask的實現(xiàn)就是發(fā)起了一個定時任務(wù)：

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

定時任務(wù)延時1秒執(zhí)行。

跟蹤到這里就告一階段了。核心功能只有兩個：調(diào)用訂閱方法和發(fā)起定時任務(wù)。

定時任務(wù)都干了啥

UpdateTask封裝了訂閱機(jī)制的核心業(yè)務(wù)邏輯，先來通過一張流程圖看一下都做了啥。

nacos

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

 
 
 
 
    
  
  
  
  
  1. public void run() { 
    2. 
  
  
  
  
  2.     long delayTime = DEFAULT_DELAY; 
    3. 
  
  
  
  
  3.  
    4. 
  
  
  
  
  4.     try { 
    5. 
  
  
  
  
  5.         // 判斷該注冊的Service是否被訂閱，如果沒有訂閱則不再執(zhí)行 
    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.             // 根據(jù)serviceName從注冊中心服務(wù)端獲取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.         // 過期服務(wù)（服務(wù)的最新更新時間小于等于緩存刷新時間），從注冊中心重新查詢 
    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.         // 下次更新緩存時間設(shè)置，默認(rèn)為6秒 
    35. 
  
  
  
  
  35.         // TODO multiple time can be configured. 
    36. 
  
  
  
  
  36.         delayTime = serviceObj.getCacheMillis() * DEFAULT_UPDATE_CACHE_TIME_MULTIPLE; 
    37. 
  
  
  
  
  37.         // 重置失敗數(shù)量為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.         // 下次調(diào)度刷新時間，下次執(zhí)行的時間與failCount有關(guān) 
    44. 
  
  
  
  
  44.         // failCount=0，則下次調(diào)度時間為6秒，最長為1分鐘 
    45. 
  
  
  
  
  45.         // 即當(dāng)無異常情況下緩存實例的刷新時間是6秒 
    46. 
  
  
  
  
  46.         executor.schedule(this, Math.min(delayTime << failCount, DEFAULT_DELAY * 60), TimeUnit.MILLISECONDS); 
    47. 
  
  
  
  
  47.     } 
    48. 
  
  
  
  
  48. } 
    49.

首先在判斷服務(wù)是否是被訂閱過，實現(xiàn)方法是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.

查看該方法的源碼會發(fā)現(xiàn)，這里的listenerMap正是最開始的subscribe方法中registerListener注冊的EventListener。

run方法后面的業(yè)務(wù)處理基本上都雷同了，先判斷緩存是否有ServiceInfo信息，如果沒有則查詢注冊中心、處理ServiceInfo、更新上次處理時間。

而下面判斷ServiceInfo是否失效，正是通過“上次更新時間”與當(dāng)前ServiceInfo中的“上次更新時間”做比較來判斷。如果失效，也會查詢注冊中心、處理ServiceInfo、更新上次處理時間等一系列操作。

業(yè)務(wù)邏輯最后會計算下一次定時任務(wù)的執(zhí)行時間，通過delayTime來延遲執(zhí)行。delayTime默認(rèn)為 1000L * 6，也就是6秒。而在finally里面真的發(fā)起下一次定時任務(wù)。當(dāng)出現(xiàn)異常時，下次執(zhí)行的時間與失敗次數(shù)有關(guān)，但最長不超過1分鐘。

小結(jié)

這一篇我們講了Nacos客戶端服務(wù)訂閱機(jī)制的源碼，主要有以下步驟：

第一步：訂閱方法的調(diào)用，并進(jìn)行EventListener的注冊，后面UpdateTask要用來進(jìn)行判斷;

第二步：通過委托代理類來處理訂閱邏輯，此處與獲取實例列表方法使用了同一個方法;

第三步：通過定時任務(wù)執(zhí)行UpdateTask方法，默認(rèn)執(zhí)行間隔為6秒，當(dāng)發(fā)生異常時會延長，但不超過1分鐘;

第四步：UpdateTask方法中會比較本地是否存在緩存，緩存是否過期。當(dāng)不存在或過期時，查詢注冊中心，獲取最新實例，更新最后獲取時間，處理ServiceInfo。

第五步：重新計算定時任務(wù)時間，循環(huán)執(zhí)行上述流程。


本文題目：NacosClient服務(wù)訂閱機(jī)制之核心流程
當(dāng)前URL：http://uogjgqi.cn/article/cojpogp.html
掃二維碼與項目經(jīng)理溝通

我們在微信上24小時期待你的聲音

解答本文疑問/技術(shù)咨詢/運(yùn)營咨詢/技術(shù)建議/互聯(lián)網(wǎng)交流