前言

在更新Spring Cloud Alibaba Nacos时,想到之前阅读过Apollo的源码,便在这插入记录了过来,后续更新Nacos Config源码

Apollo简介

fork 源码地址 apollo源码
参考apollo架构中心设计
主要分为 Config Service、Admin Service、Portal、Client 四部分
上文介绍到拉取消息监听器处理,本文介绍服务端如何处理客户端请求的

长轮询感知配置发布-服务端实现

让我们来看看客户端是怎么拿到消息通知的,服务端又是怎么进行长轮询将请求挂起的,下面采用一段一段的代码去解读。
GET 请求URL挂在类上 /notifications/v2,首先我们看看接口参数 :

• appId、cluster 这两个我们就不再解释了。
• notificationsAsString : 这样猜测,客户端拉取配置,何为配置,我们从上面portal跟到adminservice得知,appId+cluster+namespace即为一条发布的配置通知消息,我们猜测该参数就是一个json格式的一个或多个消息。
• dataCenter ,非必传参数,等跟踪到请求端,再看它是接受什么内容的参数

代码段一

DeferredResultWrapper deferredResultWrapper = new DeferredResultWrapper(bizConfig.longPollingTimeoutInMilli());
Set<String> namespaces = Sets.newHashSet();
Map<String, Long> clientSideNotifications = Maps.newHashMap();//key:namespace,value : 封装了messageId的对象
Map<String, ApolloConfigNotification> filteredNotifications = filterNotifications(appId, notifications);

• 初始化 DeferredResult 超时1s 默认返回 ResponseEntity<List<ApolloConfigNotification>>
• filterNotifications 方法返回 key:namespace、value:封装了messageId的对象,我们来看一下这个方法

private Map<String, ApolloConfigNotification> filterNotifications(String appId,List<ApolloConfigNotification> notifications) {
Map<String, ApolloConfigNotification> filteredNotifications = Maps.newHashMap();
for (ApolloConfigNotification notification : notifications) {
if (Strings.isNullOrEmpty(notification.getNamespaceName())) {continue;
}
//strip out .properties suffix
String originalNamespace = namespaceUtil.filterNamespaceName(notification.getNamespaceName());
notification.setNamespaceName(originalNamespace);
//fix the character case issue, such as FX.apollo <-> fx.apollo
String normalizedNamespace = namespaceUtil.normalizeNamespace(appId, originalNamespace);// in case client side namespace name has character case issue and has difference notification ids
// such as FX.apollo = 1 but fx.apollo = 2, we should let FX.apollo have the chance to update its notification id
// which means we should record FX.apollo = 1 here and ignore fx.apollo = 2
if (filteredNotifications.containsKey(normalizedNamespace) &&filteredNotifications.get(normalizedNamespace).getNotificationId() < notification.getNotificationId()) {continue;
}filteredNotifications.put(normalizedNamespace, notification);
}
return filteredNotifications;
}

• 循环遍历客户端传递过来的配置集合 留个疑问客户端传递过来的数据是怎么来的
• 处理 .properties 结尾的 namespace,即移除.properties
• normalizeNamespace 方法是从缓存获取app下的appNamespace,如果获取不到,直接返回,关于缓存的实现,可以查看 AppNamespaceServiceWithCache 类
• 接下来的一段注释翻译过来是 : 如果客户端名称空间名称具有字符大小写问题并且具有不同的通知ID,例如FX.apollo=1但FX.apollo=2，我们应该让FX.apollo有机会更新其通知id,这意味着我们应该在这里记录FX.apollo=1而忽略FX.apollo=2。有点绕,暂时不明白,放过。
• 添加到 filteredNotifications 中,循环结束后返回该Map

回到主方法,代码段二

for (Map.Entry<String, ApolloConfigNotification> notificationEntry : filteredNotifications.entrySet()) {
String normalizedNamespace = notificationEntry.getKey();
ApolloConfigNotification notification = notificationEntry.getValue();
namespaces.add(normalizedNamespace);
clientSideNotifications.put(normalizedNamespace, notification.getNotificationId());
if (!Objects.equals(notification.getNamespaceName(), normalizedNamespace)) {
deferredResultWrapper.recordNamespaceNameNormalizedResult(notification.getNamespaceName(), normalizedNamespace);
}
}

• 遍历代码段一返回的以namespace为key,以封装了messageId的对象为value的map
• 将 namespace add 进 namespaces 集合,将 namespace作为key,messageId作为value put进clientSideNotifications map
• 处理信息不对称的namespace(以客户端传递过来的为主),并记录到DeferredResult 中

代码段三

//key:namespace、value:appId+cluster+namespace
Multimap<String, String> watchedKeysMap =
watchKeysUtil.assembleAllWatchKeys(appId, cluster, namespaces, dataCenter);
//将values set 进集合
Set<String> watchedKeys = Sets.newHashSet(watchedKeysMap.values());

我们进入 assembleAllWatchKeys 方法

/**
* Assemble watch keys for the given appId, cluster, namespaces, dataCenter combination
* 译文 : 为给定的appId、集群、命名空间、数据中心组合装配监视键
* @return a multimap with namespace as the key and watch keys as the value
*       译文:以命名空间为键、以监视键为值的多重映射
*/
public Multimap<String, String> assembleAllWatchKeys(String appId, String clusterName,Set<String> namespaces,String dataCenter) {
Multimap<String, String> watchedKeysMap =assembleWatchKeys(appId, clusterName, namespaces, dataCenter);//Every app has an 'application' namespace
if (!(namespaces.size() == 1 && namespaces.contains(ConfigConsts.NAMESPACE_APPLICATION))) {
Set<String> namespacesBelongToAppId = namespacesBelongToAppId(appId, namespaces);
Set<String> publicNamespaces = Sets.difference(namespaces, namespacesBelongToAppId);//Listen on more namespaces if it's a public namespace
if (!publicNamespaces.isEmpty()) {watchedKeysMap.putAll(findPublicConfigWatchKeys(appId, clusterName, publicNamespaces, dataCenter));
}
}return watchedKeysMap;
}

assembleWatchKeys 方法

private Set<String> assembleWatchKeys(String appId, String clusterName, String namespace,String dataCenter) {
//ApolloNoAppIdPlaceHolder
if (ConfigConsts.NO_APPID_PLACEHOLDER.equalsIgnoreCase(appId)) {
return Collections.emptySet();
}
Set<String> watchedKeys = Sets.newHashSet();//监听非default的集群
//watch specified cluster config change
if (!Objects.equals(ConfigConsts.CLUSTER_NAME_DEFAULT, clusterName)) {
watchedKeys.add(assembleKey(appId, clusterName, namespace));
}//watch data center config change
if (!Strings.isNullOrEmpty(dataCenter) && !Objects.equals(dataCenter, clusterName)) {
watchedKeys.add(assembleKey(appId, dataCenter, namespace));
}//watch default cluster config change
watchedKeys.add(assembleKey(appId, ConfigConsts.CLUSTER_NAME_DEFAULT, namespace));return watchedKeys;
}

由此可知,default集群的配置一定会加入监听,返回的map key 为 appId+clusterName+namespace

返回到外层,继续往下看 assembleAllWatchKeys 方法

if (!(namespaces.size() == 1 && namespaces.contains(ConfigConsts.NAMESPACE_APPLICATION)))

这是一个或语句,size != 1 或 不是监听 application 集群,就处理一些逻辑,暂且我们略过。

代码段四

/**
* 1、set deferredResult before the check, for avoid more waiting
* If the check before setting deferredResult,it may receive a notification the next time
* when method handleMessage is executed between check and set deferredResult.
* 译文 : 再检查之前设置deferredResult，以避免更多的等待,如果在设置deferredResult之前进行检查，则下次可能会收到通知
*       方法handleMessage在check和set deferredResult之间执行时。
*/
deferredResultWrapper.onTimeout(() -> logWatchedKeys(watchedKeys, "Apollo.LongPoll.TimeOutKeys"));deferredResultWrapper.onCompletion(() -> {
//unregister all keys
for (String key : watchedKeys) {
deferredResults.remove(key, deferredResultWrapper);
}
logWatchedKeys(watchedKeys, "Apollo.LongPoll.CompletedKeys");
});//register all keys
for (String key : watchedKeys) {
this.deferredResults.put(key, deferredResultWrapper);
}logWatchedKeys(watchedKeys, "Apollo.LongPoll.RegisteredKeys");
logger.debug("Listening {} from appId: {}, cluster: {}, namespace: {}, datacenter: {}",
watchedKeys, appId, cluster, namespaces, dataCenter);

• 先设置 DeferredResult 超时处理,前面初始化设置的是60s,也就是说,在 handleMessage setResult超时会来回调 onTimeout 方法,记录 watchedKeys的Tracer日志
• onCompletion 方法,会在 handleMessage 方法 setResult后触发,清除 deferredResults ,这个 deferredResults 我们在 handleMessage 方法中也看到过它:

if (!deferredResults.containsKey(content)) {
return;
}//create a new list to avoid ConcurrentModificationException
List<DeferredResultWrapper> results = Lists.newArrayList(deferredResults.get(content));

setResult回调后也会去删除它里面的元素,那它是何时初始化数据的,我们接着看

• 开始注册,deferredResults key 为: appId+cluster+namespace、value为: DeferredResult对象
• 记录长轮询key注册监听日志

代码段五

/**
* 2、check new release
*/
List<ReleaseMessage> latestReleaseMessages =
releaseMessageService.findLatestReleaseMessagesGroupByMessages(watchedKeys);

看看 ReleaseMessageServiceWithCache#findLatestReleaseMessagesGroupByMessages

public List<ReleaseMessage> findLatestReleaseMessagesGroupByMessages(Set<String> messages) {
if (CollectionUtils.isEmpty(messages)) {
return Collections.emptyList();
}
List<ReleaseMessage> releaseMessages = Lists.newArrayList();for (String message : messages) {
ReleaseMessage releaseMessage = releaseMessageCache.get(message);
if (releaseMessage != null) {releaseMessages.add(releaseMessage);
}
}return releaseMessages;
}

大概内容就是,循环遍历 message(appId+cluster+namespace) ,并从缓存中取出message对应的ReleaseMessage对象 ,若缓存存在则添加到集合并返回。

疑问:这里的缓存对象 releaseMessageCache 什么时候初始化值的? 我们看看该类实现了 ReleaseMessageListener 接口,所以它也是一个配置发布通知的类,在 handleMessage 方法里 mergeReleaseMessage 方法中, 会将刚发布接收到的新的消息put进缓存。

也就是说代码段5,是从缓存中检出最新的消息。

代码段六

/**
* Manually close the entity manager.
* Since for async request, Spring won't do so until the request is finished,
* which is unacceptable since we are doing long polling - means the db connection would be hold
* for a very long time
*/
entityManagerUtil.closeEntityManager();

/**
* close the entity manager.
* Use it with caution! This is only intended for use with async request, which Spring won't
* close the entity manager until the async request is finished.
*/
public void closeEntityManager() {
EntityManagerHolder emHolder = (EntityManagerHolder)TransactionSynchronizationManager.getResource(getEntityManagerFactory());
if (emHolder == null) {
return;
}
logger.debug("Closing JPA EntityManager in EntityManagerUtil");
EntityManagerFactoryUtils.closeEntityManager(emHolder.getEntityManager());
}

翻译过来就是:手动关闭实体管理器。因为对于异步请求，Spring在请求完成之前不会这么做，这是不可接受的，因为我们正在进行长轮询-意味着数据库连接将被保持很长一段时间。这只用于异步请求，而Spring不会关闭实体管理器，直到异步请求完成。

代码段七

List<ApolloConfigNotification> newNotifications =
getApolloConfigNotifications(namespaces, clientSideNotifications, watchedKeysMap,latestReleaseMessages);if (!CollectionUtils.isEmpty(newNotifications)) {
deferredResultWrapper.setResult(newNotifications);
}return deferredResultWrapper.getResult();

参数这里记录一下:

namespaces : 代码段二有详细介绍,namespace集合

clientSideNotifications : 代码二有详细介绍,namespace作为key,messageId为集合的map

watchedKeysMap : 代码段三有详细介绍,key为namespace,value为appId+cluster+namespace的map

latestReleaseMessages : 代码段五有详细介绍,最新发布的消息集合

private List<ApolloConfigNotification> getApolloConfigNotifications(Set<String> namespaces,Map<String, Long> clientSideNotifications,Multimap<String, String> watchedKeysMap,List<ReleaseMessage> latestReleaseMessages) {
List<ApolloConfigNotification> newNotifications = Lists.newArrayList();
if (!CollectionUtils.isEmpty(latestReleaseMessages)) {
//新建通知存储map
Map<String, Long> latestNotifications = Maps.newHashMap();
//遍历最新的消息
for (ReleaseMessage releaseMessage : latestReleaseMessages) {
//key:appId+cluster+namespace,value:消息ID
latestNotifications.put(releaseMessage.getMessage(), releaseMessage.getId());
}
//遍历namespaces
for (String namespace : namespaces) {
//获取messageId
long clientSideId = clientSideNotifications.get(namespace);
// -1
long latestId = ConfigConsts.NOTIFICATION_ID_PLACEHOLDER;//获取watchedKeys:appId+cluster+namespace
Collection<String> namespaceWatchedKeys = watchedKeysMap.get(namespace);
//遍历watchedKeys
for (String namespaceWatchedKey : namespaceWatchedKeys) {
//获取messageId,如果没有,则-1
long namespaceNotificationId =latestNotifications.getOrDefault(namespaceWatchedKey, ConfigConsts.NOTIFICATION_ID_PLACEHOLDER);
//如果 messageId > latestId
if (namespaceNotificationId > latestId) {
//重新赋值
latestId = namespaceNotificationId;
}
}
//
if (latestId > clientSideId) {
ApolloConfigNotification notification = new ApolloConfigNotification(namespace, latestId);
//namespaceWatchedKeys:appId+cluster+namespace,过滤出latestNotifications存在的key,
namespaceWatchedKeys.stream().filter(latestNotifications::containsKey).forEach(namespaceWatchedKey ->//key:appId+cluster+namespace ,value:messageIdnotification.addMessage(namespaceWatchedKey, latestNotifications.get(namespaceWatchedKey)));
//添加到通知集合
newNotifications.add(notification);
}
}
}
//返回通知集合
return newNotifications;
}

总结

• configservice通过 ReleaseMessageScnner 拉取消息,并通知所有的 ReleaseMessageListener 接口实现,告知新的消息发布,并执行handleMessage方法。
• 客户端会发送一个Http 请求到configservice的NotificationControllerV2 /notifications/v2 接口(客户端何时、何处请求上文没做讲解,下文讲解)
• 该接口不会立马返回,而是通过 DeferredResult 将请求挂起,同时设置了如果60s秒没有客户端关心的配置发布,则返回304状态码
• 如果有配置变化,调用 DeferredResult#setResult 将变化的配置消息:appId+cluster+namespace 和 messageId 返回给客户端。