源碼

Apollo 長輪詢的實現，是通過客戶端輪詢 /notifications/v2 接口實現的。具體代碼在 com.ctrip.framework.apollo.configservice.controller.NotificationControllerV2.java。

這個類也是實現了 ReleaseMessageListener 監控，表明他是一個消息監聽器，當有新的消息時，就會調用他的 hanlderMessage 方法。這個具體我們后面再說。

該類只有一個 rest 接口： pollNotification 方法。返回值是 DeferredResult，這是 Spring 支持 Servlet 3 的一個類，關于異步同步的不同，可以看筆者的另一篇文章 異步 Servlet 和同步 Servlet 的性能測試。

該接口提供了幾個參數：

appId appId

cluster 集群名稱

notificationsAsString 通知對象的 json 字符串

dataCenter，idc 屬性

clientIp 客戶端 IP， 非必傳，為了擴展吧估計

大家有么有覺得少了什么？ namespace 。

當然，沒有 namespace 這個重要的參數是不存在的。

參數在 notificationsAsString 中。客戶端會將自己所有的 namespace 傳遞到服務端進行查詢。

是時候上源碼了。

@RequestMapping(method = RequestMethod.GET)public DeferredResult<ResponseEntity<List<ApolloConfigNotification>>> pollNotification(@RequestParam(value = "appId") String appId,// appId@RequestParam(value = "cluster") String cluster,// default@RequestParam(value = "notifications") String notificationsAsString,// json 對象 List<ApolloConfigNotification>@RequestParam(value = "dataCenter", required = false) String dataCenter,// 基本用不上, idc 屬性@RequestParam(value = "ip", required = false) String clientIp) {List<ApolloConfigNotification> notifications =// 轉換成對象gson.fromJson(notificationsAsString, notificationsTypeReference);// Spring 的異步對象: timeout 60s, 返回304DeferredResultWrapper deferredResultWrapper = new DeferredResultWrapper();Set<String> namespaces = Sets.newHashSet();Map<String, Long> clientSideNotifications = Maps.newHashMap();Map<String, ApolloConfigNotification> filteredNotifications = filterNotifications(appId, notifications);// 過濾一下名字// 循環for (Map.Entry<String, ApolloConfigNotification> notificationEntry : filteredNotifications.entrySet()) {// 拿出 keyString normalizedNamespace = notificationEntry.getKey();// 拿出 valueApolloConfigNotification notification = notificationEntry.getValue();/* 添加到 namespaces Set */namespaces.add(normalizedNamespace);// 添加到 client 端的通知, key 是 namespace, values 是 messageIdclientSideNotifications.put(normalizedNamespace, notification.getNotificationId());// 如果不相等, 記錄客戶端名字if (!Objects.equals(notification.getNamespaceName(), normalizedNamespace)) {// 記錄 key = 標準名字, value = 客戶端名字deferredResultWrapper.recordNamespaceNameNormalizedResult(notification.getNamespaceName(), normalizedNamespace);}}// 記在 namespaces 集合, clientSideNotifications 也put (namespace, notificationId)// 組裝得到需要觀察的 key,包括公共的.Multimap<String, String> watchedKeysMap =watchKeysUtil.assembleAllWatchKeys(appId, cluster, namespaces, dataCenter);// namespaces 是集合// 得到 value; 這個 value 也就是 appId + cluster + namespaceSet<String> watchedKeys = Sets.newHashSet(watchedKeysMap.values());// 從緩存得到最新的發布消息List<ReleaseMessage> latestReleaseMessages =// 根據 key 從緩存得到最新發布的消息.releaseMessageService.findLatestReleaseMessagesGroupByMessages(watchedKeys);/* 如果不關閉, 這個請求將會一直持有一個數據庫連接. 影響并發能力. 這是一個 hack 操作*/entityManagerUtil.closeEntityManager();// 計算出新的通知List<ApolloConfigNotification> newNotifications =getApolloConfigNotifications(namespaces, clientSideNotifications, watchedKeysMap,latestReleaseMessages);// 不是空, 理解返回結果, 不等待if (!CollectionUtils.isEmpty(newNotifications)) {deferredResultWrapper.setResult(newNotifications);} else {// 設置 timeout 回調:打印日志deferredResultWrapper.onTimeout(() -> logWatchedKeys(watchedKeys, "Apollo.LongPoll.TimeOutKeys"));// 設置完成回調:刪除 keydeferredResultWrapper.onCompletion(() -> {//取消注冊for (String key : watchedKeys) {deferredResults.remove(key, deferredResultWrapper);}});//register all keys 注冊for (String key : watchedKeys) {this.deferredResults.put(key, deferredResultWrapper);}}// 立即返回return deferredResultWrapper.getResult();/** @see DeferredResultHandler 是關鍵 */}

注釋寫了很多了，再簡單說說邏輯：

解析 JSON 字符串為 List< ApolloConfigNotification> 對象。

創建 Spring 異步對象。

處理過濾 namespace。

根據 namespace 生成需要監聽的 key，格式為 appId + cluster + namespace，包括公共 namespace。并獲取最新的 Release 信息。

關閉 Spring 實例管理器，釋放數據庫資源。

根據剛剛得到的 ReleaseMessage，和客戶端的 ReleaseMessage 的版本進行對比，生成新的配置通知對象集合。

如果不是空 —— 立即返回給客戶端，結束此次調用。如果沒有，進入第 8 步。

設置 timeout 回調方法 —— 打印日志。再設置完成回調方法：刪除注冊的 key。

對客戶端感興趣的 key 進行注冊，這些 key 都對應著 deferredResultWrapper 對象，可以認為他就是客戶端。

返回 Spring 異步對象。該請求將被異步掛起。

Apollo 的 DeferredResultWrapper 保證了 Spring 的 DeferredResult 對象，泛型內容是 List， 構造這個對象，默認的 timeout 是 60 秒，即掛起 60 秒。同時，對 setResult 方法進行包裝，加入了對客戶端 key 和服務端 key 的一個映射（大小寫不一致） 。

我們剛剛說，Apollo 會將這些 key 注冊起來。那么什么時候使用呢，異步對象被掛起，又是上面時候被喚醒呢？

答案就在 handleMessage 方法里。我們剛剛說他是一個監聽器，當消息掃描器掃描到新的消息時，會通知所有的監聽器，也就是執行 handlerMessage 方法。方法內容如下：

@Override public void handleMessage(ReleaseMessage message, String channel) {String content = message.getMessage();if (!Topics.APOLLO_RELEASE_TOPIC.equals(channel) || Strings.isNullOrEmpty(content)) {return;}String changedNamespace = retrieveNamespaceFromReleaseMessage.apply(content);//create a new list to avoid ConcurrentModificationException 構造一個新 list ,防止并發失敗List<DeferredResultWrapper> results = Lists.newArrayList(deferredResults.get(content));// 創建通知對象ApolloConfigNotification configNotification = new ApolloConfigNotification(changedNamespace, message.getId());configNotification.addMessage(content, message.getId());//do async notification if too many clients 如果有大量的客戶端(100)在等待,使用線程池異步處理if (results.size() > bizConfig.releaseMessageNotificationBatch()) {// 大量通知批量處理largeNotificationBatchExecutorService.submit(() -> {for (int i = 0; i < results.size(); i++) { // 循環/** 假設一個公共 Namespace 有10W 臺機器使用，如果該公共 Namespace 發布時直接下發配置更新消息的話，* 就會導致這 10W 臺機器一下子都來請求配置，這動靜就有點大了，而且對 Config Service 的壓力也會比較大。* 即"驚群效應"*/if (i > 0 && i % bizConfig.releaseMessageNotificationBatch() == 0) {// 如果處理了一批客戶端,休息一下(100ms)TimeUnit.MILLISECONDS.sleep(bizConfig.releaseMessageNotificationBatchIntervalInMilli());}results.get(i).setResult(configNotification);// 通知每個等待的 HTTP 請求}});return;}// 否則,同步處理for (DeferredResultWrapper result : results) {result.setResult(configNotification);} }

筆者去除了一些日志和一些數據判斷。大致的邏輯如下：

消息類型必須是 “apollo-release”。然后拿到消息里的 namespace 內容。

根據 namespace 從注冊器里拿出 Spring 異步對象集合。

創建通知對象。

如果有超過 100 個客戶端在等待，那么就使用線程池批量執行通知。否則就同步慢慢執行。

每處理 100 個客戶端就休息 100ms，防止發生驚群效應，導致大量客戶端調用配置獲取接口，引起服務抖動。

循環調用 Spring 異步對象的 setResult 方法，讓其立即返回。

具體的流程圖如下：

其中，灰色區域是掃描器的異步線程，黃色區域是接口的同步線程。他們共享 deferredResults 這個線程安全的 Map，實現異步解耦和實時通知客戶端。

總結

好了，這就是 Apollo 的長輪詢接口，客戶端會不斷的輪詢服務器，服務器會 Hold住 60 秒，這是通過 Servlet 3 的異步 + NIO 來實現的，能夠保持萬級連接（Tomcat 默認 10000）。

通過一個線程安全的 Map + 監聽器，讓掃描器線程和 HTTP 線程共享 Spring 異步對象，即實現了消息實時通知，也讓應用程序實現異步解耦。

轉載于:https://www.cnblogs.com/stateis0/p/9393871.html

總結

以上是生活随笔為你收集整理的Apollo 8 — ConfigService 异步轮询接口的实现的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。