黑马程序员技术交流社区

标题: 【上海校区】配置中心 Apollo 源码解析 —— Config Service 通... [打印本页]
作者: 不二晨    时间: 2018-11-9 09:31
标题: 【上海校区】配置中心 Apollo 源码解析 —— Config Service 通...
1. 概述
老艿艿：本系列假定胖友已经阅读过 《Apollo 官方 wiki 文档》 。
本文接 《Apollo 源码解析 —— Admin Service 发送 ReleaseMessage》 一文，分享配置发布的第四步，NotificationControllerV2 得到配置发布的 AppId+Cluster+Namespace 后，会通知对应的客户端 。
FROM 《Apollo配置中心设计》 的 2.1.2 Config Service 通知客户端的实现方式
友情提示：在目前 Apollo 的实现里，如下的名词是“等价”的：
文章暂时未统一用词，所以胖友看的时候需要“脑补”下。
2. NotificationControllerV2
老艿艿：流程较长，代码较多，请耐心理解。
com.ctrip.framework.apollo.configservice.controller.NotificationControllerV2 ，实现 ReleaseMessageListener 接口，通知 Controller ，仅提供 notifications/v2 接口。
2.1 构造方法
/** * Watch Key 与 DeferredResultWrapper 的 Multimap * * Key：Watch Key * Value：DeferredResultWrapper 数组 */private final Multimap<String, DeferredResultWrapper> deferredResults = Multimaps.synchronizedSetMultimap(HashMultimap.create());private static final Splitter STRING_SPLITTER = Splitter.on(ConfigConsts.CLUSTER_NAMESPACE_SEPARATOR).omitEmptyStrings();private static final Type notificationsTypeReference = new TypeToken<List<ApolloConfigNotification>>() {}.getType();/** * 通过 ReleaseMessage 的消息内容，获得对应 Namespace 的名字 */private static final Function<String, String> retrieveNamespaceFromReleaseMessage =        releaseMessage -> {            if (Strings.isNullOrEmpty(releaseMessage)) {                return null;            }            List<String> keys = STRING_SPLITTER.splitToList(releaseMessage);            //message should be appId+cluster+namespace            if (keys.size() != 3) {                logger.error("message format invalid - {}", releaseMessage);                return null;            }            return keys.get(2);        };/** * 大量通知分批执行 ExecutorService */private final ExecutorService largeNotificationBatchExecutorService;@Autowiredprivate WatchKeysUtil watchKeysUtil;@Autowiredprivate ReleaseMessageServiceWithCache releaseMessageService;@Autowiredprivate EntityManagerUtil entityManagerUtil;@Autowiredprivate NamespaceUtil namespaceUtil;@Autowiredprivate Gson gson;@Autowiredprivate BizConfig bizConfig;public NotificationControllerV2() {    largeNotificationBatchExecutorService = Executors.newSingleThreadExecutor(ApolloThreadFactory.create("NotificationControllerV2", true));}
2.2 pollNotification
1: @RequestMapping(method = RequestMethod.GET) 2: public DeferredResult<ResponseEntity<List<ApolloConfigNotification>>> pollNotification( 3:         @RequestParam(value = "appId") String appId, 4:         @RequestParam(value = "cluster") String cluster, 5:         @RequestParam(value = "notifications") String notificationsAsString, 6:         @RequestParam(value = "dataCenter", required = false) String dataCenter, 7:         @RequestParam(value = "ip", required = false) String clientIp) { 8:     // 解析 notificationsAsString 参数，创建 ApolloConfigNotification 数组。 9:     List<ApolloConfigNotification> notifications = null;10:     try {11:         notifications = gson.fromJson(notificationsAsString, notificationsTypeReference);12:     } catch (Throwable ex) {13:         Tracer.logError(ex);14:     }15:     if (CollectionUtils.isEmpty(notifications)) {16:         throw new BadRequestException("Invalid format of notifications: " + notificationsAsString);17:     }18:19:     // 创建 DeferredResultWrapper 对象20:     DeferredResultWrapper deferredResultWrapper = new DeferredResultWrapper();21:     // Namespace 集合22:     Set<String> namespaces = Sets.newHashSet();23:     // 客户端的通知 Map 。key 为 Namespace 名，value 为通知编号。24:     Map<String, Long> clientSideNotifications = Maps.newHashMap();25:     // 过滤并创建 ApolloConfigNotification Map26:     Map<String, ApolloConfigNotification> filteredNotifications = filterNotifications(appId, notifications);27:     // 循环 ApolloConfigNotification Map ，初始化上述变量。28:     for (Map.Entry<String, ApolloConfigNotification> notificationEntry : filteredNotifications.entrySet()) {29:         String normalizedNamespace = notificationEntry.getKey();30:         ApolloConfigNotification notification = notificationEntry.getValue();31:         // 添加到 `namespaces` 中。32:         namespaces.add(normalizedNamespace);33:         // 添加到 `clientSideNotifications` 中。34:         clientSideNotifications.put(normalizedNamespace, notification.getNotificationId());35:         // 记录名字被归一化的 Namespace 。因为，最终返回给客户端，使用原始的 Namespace 名字，否则客户端无法识别。36:         if (!Objects.equals(notification.getNamespaceName(), normalizedNamespace)) {37:             deferredResultWrapper.recordNamespaceNameNormalizedResult(notification.getNamespaceName(), normalizedNamespace);38:         }39:     }40:     if (CollectionUtils.isEmpty(namespaces)) {41:         throw new BadRequestException("Invalid format of notifications: " + notificationsAsString);42:     }43:44:     // 组装 Watch Key Multimap45:     Multimap<String, String> watchedKeysMap = watchKeysUtil.assembleAllWatchKeys(appId, cluster, namespaces, dataCenter);46:     // 生成 Watch Key 集合47:     Set<String> watchedKeys = Sets.newHashSet(watchedKeysMap.values());48:     // 获得 Watch Key 集合中，每个 Watch Key 对应的 ReleaseMessage 记录。49:     List<ReleaseMessage> latestReleaseMessages = releaseMessageService.findLatestReleaseMessagesGroupByMessages(watchedKeys);50:51:     /**52:      * Manually close the entity manager.53:      * Since for async request, Spring won't do so until the request is finished,54:      * which is unacceptable since we are doing long polling - means the db connection would be hold55:      * for a very long time56:      */57:     // 手动关闭 EntityManager58:     // 因为对于 async 请求，Spring 在请求完成之前不会这样做59:     // 这是不可接受的，因为我们正在做长轮询——意味着 db 连接将被保留很长时间。60:     // 实际上，下面的过程，我们已经不需要 db 连接，因此进行关闭。61:     entityManagerUtil.closeEntityManager();62:     // 获得新的 ApolloConfigNotification 通知数组63:     List<ApolloConfigNotification> newNotifications = getApolloConfigNotifications(namespaces, clientSideNotifications, watchedKeysMap, latestReleaseMessages);64:     // 若有新的通知，直接设置结果。65:     if (!CollectionUtils.isEmpty(newNotifications)) {66:         deferredResultWrapper.setResult(newNotifications);67:         // 若无新的通知，68:     } else {69:         // 注册超时事件70:         deferredResultWrapper.onTimeout(() -> logWatchedKeys(watchedKeys, "Apollo.LongPoll.TimeOutKeys")); // 【TODO 6001】Tracer 日志71:         // 注册结束事件72:         deferredResultWrapper.onCompletion(() -> {73:             // 移除 Watch Key + DeferredResultWrapper 出 `deferredResults`74:             // unregister all keys75:             for (String key : watchedKeys) {76:                 deferredResults.remove(key, deferredResultWrapper);77:             }78:             // 【TODO 6001】Tracer 日志79:             logWatchedKeys(watchedKeys, "Apollo.LongPoll.CompletedKeys");80:         });81:82:         // 注册 Watch Key + DeferredResultWrapper 到 `deferredResults` 中，等待配置发生变化后通知。详见 `#handleMessage(...)` 方法。83:         // register all keys84:         for (String key : watchedKeys) {85:             this.deferredResults.put(key, deferredResultWrapper);86:         }87:88:         // 【TODO 6001】Tracer 日志89:         logWatchedKeys(watchedKeys, "Apollo.LongPoll.RegisteredKeys");90:         logger.debug("Listening {} from appId: {}, cluster: {}, namespace: {}, datacenter: {}", watchedKeys, appId, cluster, namespaces, dataCenter);91:     }92:93:     return deferredResultWrapper.getResult();94: }