Apollo 7 — ConfigService 消息扫描设计实现
目录
- 设计
- 代码实现
- 总结
1.设计
Apollo 为了减少依赖，将本来 MQ 的职责转移到了 Mysql 中。具体表现为 Mysql 中的 ReleaseMessage 表。
具体官方文档可见：发送ReleaseMessage的实现方式
用张图简单的来表示一下 ：
有人肯定要问了，为什么 Admin Service 和 Config Service 不放在一起呢？我曾提过 issue 问过作者，大概的答案是：两者职责不同，部署的实例数也不同，通常 Admin 会少一些，因为只服务于 Portal，而 Config 则要部署的多一些，因为需要服务于 Client。
第二则是两者的开发节奏也是不一样，Config Service 的更新会影响客户端，而 Admin 的更新则是影响 Portal，所以，分开他们，对服务的部署可以更加细粒度。
关于这个 issue：为什么 admin 和 config 不合在一起呢
高可用的话，可参考下图或链接 可用性考虑:
扯远了。
回到我们之前说的，之所以要使用 Mysql，是因为要减少依赖，为了替代 MQ，而之所以要使用 MQ，是为了解耦 Config 和 Admin，而之所以要使用 Config 和 Admin，则是因为设计，部署，开发节奏等原因。
那么，基于 Mysql 的消息的实现是怎么弄的呢？
上面是 apollo 文档对于发送ReleaseMessage的实现方式的描述。
我们来看看代码实现.
2. 代码实现
在 com.ctrip.framework.apollo.biz.message 包下，有关于消息的实现，
麻雀虽小，五脏俱全。
-
MessageSender 接口定义了一个方法：
void sendMessage(String message, String channel);
这个 channel 就是 topic 了。 message 就是消息的具体内容了。 -
目前只有一个实现类
DatabaseMessageSender， 基于数据库的消息发送。就是把消息保存到 ReleaseMessage 表中。 -
Topics 定义消息主题，目前只有一个：
apollo-release。 -
ReleaseMessageListener 消息监听器接口，定义了一个方法：
void handleMessage(ReleaseMessage message, String channel)，需要实现此方法并注册到扫描器中，当有新的消息时，便会通知监听器。 -
ReleaseMessageScanner 消息扫描器，用于扫描数据库的 ReleaseMessage 表，如果有新数据，则通知监听器。
目前监听器有以下实现：
从左到右：
- ReleaseMessageServiceWithCache，ReleaseMessage 的缓存，用于长轮询判断是否以后新的消息。
- GrayReleaseRulesHolder，灰度规则变化监听器。
- ConfigService，分为缓存和默认，默认的 handleMessage 方法什么都不做，缓存实现则会对 cache 热身。
- NotificationControllerV2 监听器，当客户端被长连接 Hold 住时，消息如果更新则唤醒客户端立即返回，保证及时性。
- NotificationController 已废弃，不谈了。
所以，每当 ReleaseMessageScanner 得到新的消息，都会触发这些监听器。这些监听器在哪里被添加的呢？
位置:com.ctrip.framework.apollo.configservice.ConfigServiceAutoConfiguration.java
ReleaseMessageScanner 的 afterPropertiesSet 方法会启动一个间隔 1 秒定时任务，执行 scanMessages 方法。
具体方法如下：
private boolean scanAndSendMessages() {
//current batch is 500 批处理 500 条
// 根据 maxIdScanned 找到比这个 id 大的 500 条数据,
List<ReleaseMessage> releaseMessages =
releaseMessageRepository.findFirst500ByIdGreaterThanOrderByIdAsc(maxIdScanned);
if (CollectionUtils.isEmpty(releaseMessages)) {
return false;
}
// 开始通知 handleMessage 监听器
fireMessageScanned(releaseMessages);
int messageScanned = releaseMessages.size();// 消息数量
maxIdScanned = releaseMessages.get(messageScanned - 1).getId();// 更新最大 id
return messageScanned == 500;// 如果不足 500, 说明没有新消息了
}方法很简单，首先根据最大的扫描 Id 找到 500 条消息，对这 500 条消息进行批处理，触发监听器。最后更新最大扫描 Id。如果此次取出的数据量超过 500 条，则认为还有数据，就继续处理。
很明显，每次处理这么多消息肯定很耗时，那么假设处理这些消息要 5 秒，那么定时任务的间隔是多少了呢？答：6 秒。因为定时任务的模式是 scheduleWithFixedDelay 模式，固定的间隔，以当前任务结束时间 + period 时间作为下一个任务的开始时间。
那么，admin 什么时候会向数据库发送消息（保存 ReleaseMessage）呢?
那么就要查看 sendMessage 方法被哪些地方调用就好了。
-
NamespaceBranchController
1.1 更新灰度规则updateBranchGrayRules
1.2 删除灰度分支deleteBranch -
NamespaceService
2.1 删除命名空间NamespaceController#deleteNamespace
2.2 删除集群ClusterController#delete
2.3 删除灰度NamespaceBranchController#deleteBranch
2.4 删除命名空间分支NamespaceService#deleteNamespace -
ReleaseController
3.1 主/灰版本发布publish
3.2 全量发布updateAndPublish
3.3 回滚rollback
这些地方被触发的时候，都会发送消息到数据库。因此，可能会重复触发，导致 ConfigService 重复消费。。。例如在放弃灰度和全量发布的时候，就会重复发送消息。
笔者就这个问题提了 issue，期待官方 fix 这个 bug。
3. 总结
本文重点分析了 apollo 关系消息这块的设计： 每当有 app， cluster，namespace， Release 等操作的时候，都会发送消息到数据库，ConfigService 会定时扫描数据库，有新消息了，就会立即通知各个监听器，确保配置实时推送到客户端。
同时，我们也发现了一个 bug，就是会重复发送消息，引起重复消费。
迫于种种限制，apollo 使用了这种设计，获取，如果可以使用 MQ 的话，一切会更加的简单。