【Apollo】【三】发布配置的过程-变更配置消息的发送与消费-526互联

1 前言

上节我们看了下发布配置的 Portal 以及 Admin Service的变化过程，我们看到Admin Service 的 messageSender.sendMessage()，发送消息以及消费消息，那么这节我们继续看。

2 MessageSender 发送消息

2.1 ReleaseMessage

我们首先回顾下 ReleaseMessage 的结构哈，com.ctrip.framework.apollo.biz.entity.ReleaseMessage ，不继承 BaseEntity 抽象类，ReleaseMessage 实体。代码如下：

@Entity
@Table(name = "ReleaseMessage")
public class ReleaseMessage {

    /**
     * 编号
     */
    @Id
    @GeneratedValue
    @Column(name = "Id")
    private long id;
    /**
     * 消息内容，通过 {@link com.ctrip.framework.apollo.biz.utils.ReleaseMessageKeyGenerator#generate(String, String, String)} 方法生成。
     */
    @Column(name = "Message", nullable = false)
    private String message;
    /**
     * 最后更新时间
     */
    @Column(name = "DataChange_LastTime")
    private Date dataChangeLastModifiedTime;
 
    @PrePersist
    protected void prePersist() {
        if (this.dataChangeLastModifiedTime == null) {
            dataChangeLastModifiedTime = new Date();
        }
    }   
}

id 字段，编号，自增。
message 字段，消息内容。通过 ReleaseMessageKeyGenerator 生成。
#dataChangeLastModifiedTime 字段，最后更新时间。
- #prePersist() 方法，若保存时，未设置该字段，进行补全。

2.1.1 ReleaseMessageKeyGenerator

com.ctrip.framework.apollo.biz.utils.ReleaseMessageKeyGenerator ，ReleaseMessage 消息内容( ReleaseMessage.message )生成器。代码如下：

public class ReleaseMessageKeyGenerator {

    private static final Joiner STRING_JOINER = Joiner.on(ConfigConsts.CLUSTER_NAMESPACE_SEPARATOR);

    public static String generate(String appId, String cluster, String namespace) {
        return STRING_JOINER.join(appId, cluster, namespace);
    }

}

#generate(...) 方法，将 appId + cluster + namespace 拼接，使用 ConfigConsts.CLUSTER_NAMESPACE_SEPARATOR = "+" 作为间隔，例如："test+default+application" 。

因此，对于同一个 Namespace ，生成的消息内容是相同的。通过这样的方式，我们可以使用最新的 ReleaseMessage 的 id 属性，作为 Namespace 是否发生变更的标识。而 Apollo 确实是通过这样的方式实现，Client 通过不断使用获得到 ReleaseMessage 的 id 属性作为版本号，请求 Config Service 判断是否配置发生了变化。

正因为，ReleaseMessage 设计的意图是作为配置发生变化的通知，所以对于同一个 Namespace ，仅需要保留其最新的 ReleaseMessage 记录即可。所以，在下边的 DatabaseMessageSender 类中我们会看到，有后台任务不断清理旧的 ReleaseMessage 记录。

2.1.2 ReleaseMessageRepository

com.ctrip.framework.apollo.biz.repository.ReleaseMessageRepository ，继承 org.springframework.data.repository.PagingAndSortingRepository 接口，提供 ReleaseMessage 的数据访问给 Admin Service 和 Config Service 。代码如下：

public interface ReleaseMessageRepository extends PagingAndSortingRepository<ReleaseMessage, Long> {

    List<ReleaseMessage> findFirst500ByIdGreaterThanOrderByIdAsc(Long id);

    ReleaseMessage findTopByOrderByIdDesc();

    ReleaseMessage findTopByMessageInOrderByIdDesc(Collection<String> messages);

    List<ReleaseMessage> findFirst100ByMessageAndIdLessThanOrderByIdAsc(String message, Long id);

    @Query("select message, max(id) as id from ReleaseMessage where message in :messages group by message")
    List<Object[]> findLatestReleaseMessagesGroupByMessages(@Param("messages") Collection<String> messages);

}

2.2 发送入口

在上节我们的发布 ReleaseController 的 #publish(...) 方法中，会调用 MessageSender#sendMessage(message, channel) 方法，发送 Message 。调用简化代码如下：

// send release message
// 获得 Cluster 名
Namespace parentNamespace = namespaceService.findParentNamespace(namespace);
String messageCluster;
if (parentNamespace != null) { //  有父 Namespace ，说明是灰度发布，使用父 Namespace 的集群名
    messageCluster = parentNamespace.getClusterName();
} else {
    messageCluster = clusterName; // 使用请求的 ClusterName
}
// 发送 Release 消息
messageSender.sendMessage(ReleaseMessageKeyGenerator.generate(appId, messageCluster, namespaceName), Topics.APOLLO_RELEASE_TOPIC);

关于父 Namespace 部分的代码，跟灰度发布的内容相关，可以先不理解。
ReleaseMessageKeyGenerator#generate(appId, clusterName, namespaceName) 方法，生成 ReleaseMessage 的消息内容。
使用 Topic 为 Topics.APOLLO_RELEASE_TOPIC 。

看命名，数据库的方式实现消息发送的，看人家这扩展性留的，抽象以及实现，挺好，我们写业务的代码也要这样，一个点一个动作的扩展性哈。

简单看下 com.ctrip.framework.apollo.biz.message.MessageSender ，Message 发送者接口。代码如下：

public interface MessageSender {

    /**
     * 发送 Message
     *
     * @param message 消息
     * @param channel 通道（主题）
     */
    void sendMessage(String message, String channel);

}

2.2.1 Topics

com.ctrip.framework.apollo.biz.message.Topics ，Topic 枚举。代码如下：

public class Topics {

    /**
     * Apollo 配置发布 Topic
     */
    public static final String APOLLO_RELEASE_TOPIC = "apollo-release";

}

2.2.2 DatabaseMessageSender

com.ctrip.framework.apollo.biz.message.DatabaseMessageSender ，实现 MessageSender 接口，Message 发送者实现类，基于数据库实现。

2.2.2.1 构造方法

/**
 * 清理 Message 队列 最大容量
 */
private static final int CLEAN_QUEUE_MAX_SIZE = 100;

/**
 * 清理 Message 队列
 */
private BlockingQueue<Long> toClean = Queues.newLinkedBlockingQueue(CLEAN_QUEUE_MAX_SIZE);
/**
 * 清理 Message ExecutorService
 */
private final ExecutorService cleanExecutorService;
/**
 * 是否停止清理 Message 标识
 */
private final AtomicBoolean cleanStopped;

@Autowired
private ReleaseMessageRepository releaseMessageRepository;

public DatabaseMessageSender() {
    // 创建 ExecutorService 对象
    cleanExecutorService = Executors.newSingleThreadExecutor(ApolloThreadFactory.create("DatabaseMessageSender", true));
    // 设置 cleanStopped 为 false 
    cleanStopped = new AtomicBoolean(false);
}

主要和清理 ReleaseMessage 相关的属性。

2.2.2.2 sendMessage

 1 @Override
 2 @Transactional
 3 public void sendMessage(String message, String channel) {
 4     logger.info("Sending message {} to channel {}", message, channel);
 5     // 仅允许发送 APOLLO_RELEASE_TOPIC
 6     if (!Objects.equals(channel, Topics.APOLLO_RELEASE_TOPIC)) {
 7         logger.warn("Channel {} not supported by DatabaseMessageSender!");
 8         return;
 9     }
10     // 【TODO 6001】Tracer 日志
11     Tracer.logEvent("Apollo.AdminService.ReleaseMessage", message);
12     // 【TODO 6001】Tracer 日志
13     Transaction transaction = Tracer.newTransaction("Apollo.AdminService", "sendMessage");
14     try {
15         // 保存 ReleaseMessage 对象
16         ReleaseMessage newMessage = releaseMessageRepository.save(new ReleaseMessage(message));
17         // 添加到清理 Message 队列。若队列已满，添加失败，不阻塞等待。
18         toClean.offer(newMessage.getId());
19         // 【TODO 6001】Tracer 日志
20         transaction.setStatus(Transaction.SUCCESS);
21     } catch (Throwable ex) {
22         // 【TODO 6001】Tracer 日志
23         logger.error("Sending message to database failed", ex);
24         transaction.setStatus(ex);
25         throw ex;
26     } finally {
27         // 【TODO 6001】Tracer 日志
28         transaction.complete();
29     }
30 }

第 5 至 9 行：第 5 至 9 行：仅允许发送 APOLLO_RELEASE_TOPIC 。
第 16 行：调用 ReleaseMessageRepository#save(ReleaseMessage) 方法，保存 ReleaseMessage 对象。
第 18 行：调用 toClean#offer(Long id) 方法，添加到清理 Message 队列。若队列已满，添加失败，不阻塞等待。

2.2.2.3 清理 ReleaseMessage 任务

#initialize() 方法，@PostConstruct 通知 Spring 调用，初始化清理 ReleaseMessage 任务。代码如下：

 1 @PostConstruct
 2 private void initialize() {
 3     cleanExecutorService.submit(() -> {
 4         // 若未停止，持续运行。
 5         while (!cleanStopped.get() && !Thread.currentThread().isInterrupted()) {
 6             try {
 7                 // 拉取
 8                 Long rm = toClean.poll(1, TimeUnit.SECONDS);
 9                 // 队列非空，处理拉取到的消息
10                 if (rm != null) {
11                     cleanMessage(rm);
12                 // 队列为空，sleep ，避免空跑，占用 CPU
13                 } else {
14                     TimeUnit.SECONDS.sleep(5);
15                 }
16             } catch (Throwable ex) {
17                 // 【TODO 6001】Tracer 日志
18                 Tracer.logError(ex);
19             }
20         }
21     });
22 }

第 3 至 21 行：调用 ExecutorService#submit(Runnable) 方法，提交清理 ReleaseMessage 任务
- 第 5 行：循环，直到停止。
- 第 8 行：调用 BlockingQueue#poll(long timeout, TimeUnit unit) 方法，拉取队头的消息编号。
  - 第 10 至 11 行：若拉取到消息编号，调用 #cleanMessage(Long id) 方法，处理拉取到的消息，即清理老消息们。
  - 第 13 至 15 行：若未拉取到消息编号，说明队列为空，sleep ，避免空跑，占用 CPU 。

#cleanMessage(Long id) 方法，清理老消息们。代码如下：

 1 private void cleanMessage(Long id) {
 2     boolean hasMore = true;
 3     // 查询对应的 ReleaseMessage 对象，避免已经删除。因为，DatabaseMessageSender 会在多进程中执行。例如：1）Config Service + Admin Service ；2）N * Config Service ；3）N * Admin Service
 4     // double check in case the release message is rolled back
 5     ReleaseMessage releaseMessage = releaseMessageRepository.findOne(id);
 6     if (releaseMessage == null) {
 7         return;
 8     }
 9     // 循环删除相同消息内容( `message` )的老消息
10     while (hasMore && !Thread.currentThread().isInterrupted()) {
11         // 拉取相同消息内容的 100 条的老消息
12         // 老消息的定义：比当前消息编号小，即先发送的
13         // 按照 id 升序
14         List<ReleaseMessage> messages = releaseMessageRepository.findFirst100ByMessageAndIdLessThanOrderByIdAsc(
15                 releaseMessage.getMessage(), releaseMessage.getId());
16         // 删除老消息
17         releaseMessageRepository.delete(messages);
18         // 若拉取不足 100 条，说明无老消息了
19         hasMore = messages.size() == 100;
20         // 【TODO 6001】Tracer 日志
21         messages.forEach(toRemove -> Tracer.logEvent(
22                 String.format("ReleaseMessage.Clean.%s", toRemove.getMessage()), String.valueOf(toRemove.getId())));
23     }
24 }

第 5 至 8 行：调用 ReleaseMessageRepository#findOne(id) 方法，查询对应的 ReleaseMessage 对象，避免已经删除。
- 因为，DatabaseMessageSender 会在多进程中执行。例如：1）Config Service + Admin Service ；2）N * Config Service ；3）N * Admin Service 。
- 为什么 Config Service 和 Admin Service 都会启动清理任务呢？? 因为 DatabaseMessageSender 添加了 @Component 注解，而 NamespaceService 注入了 DatabaseMessageSender 。而 NamespaceService 被 apollo-adminservice 和 apoll-configservice 项目都引用了，所以都会启动该任务。
第 10 至 23 行：循环删除，相同消息内容( ReleaseMessage.message )的老消息，即 Namespace 的老消息。
- 第 14 至 15 行：调用 ReleaseMessageRepository#findFirst100ByMessageAndIdLessThanOrderByIdAsc(message, id) 方法，拉取相同消息内容的 100 条的老消息，按照 id 升序。
  - 老消息的定义：比当前消息编号小，即先发送的。
- 第 17 行：调用 ReleaseMessageRepository#delete(messages) 方法，删除老消息。
- 第 19 行：若拉取不足 100 条，说明无老消息了。
- 第 21 至 22 行：【TODO 6001】Tracer 日志

3 ReleaseMessageScanner 消费消息

到这里我们看到发布，发送消息其实是往数据库里插入了一条记录哈，那么然后呢，谁来接力呢，谁来继续呢？

消息发送完，也就是插入到数据库里谁来消费呢？我们了解下机制：

Admin Service 在配置发布后，需要通知所有的 Config Service 有配置发布，从而 Config Service 可以通知对应的客户端来拉取最新的配置。

从概念上来看，这是一个典型的消息使用场景，Admin Service 作为 producer 发出消息，各个Config Service 作为 consumer 消费消息。通过一个消息组件（Message Queue）就能很好的实现 Admin Service 和 Config Service 的解耦。

在实现上，考虑到 Apollo 的实际使用场景，以及为了尽可能减少外部依赖，我们没有采用外部的消息中间件，而是通过数据库实现了一个简单的消息队列。

实现方式如下：

Admin Service 在配置发布后会往 ReleaseMessage 表插入一条消息记录，消息内容就是配置发布的 AppId+Cluster+Namespace ，参见 DatabaseMessageSender 。
Config Service 有一个线程会每秒扫描一次 ReleaseMessage 表，看看是否有新的消息记录，参见 ReleaseMessageScanner 。
Config Service 如果发现有新的消息记录，那么就会通知到所有的消息监听器（ReleaseMessageListener），如 NotificationControllerV2 ，消息监听器的注册过程参见 ConfigServiceAutoConfiguration 。
NotificationControllerV2 得到配置发布的 AppId+Cluster+Namespace 后，会通知对应的客户端。

就是 ReleaseMessageScanner，我们继续看：

com.ctrip.framework.apollo.biz.message.ReleaseMessageScanner ，实现 org.springframework.beans.factory.InitializingBean 接口，ReleaseMessage 扫描器，被 Config Service 使用。

3.1 构造方法

@Autowired
private BizConfig bizConfig;
@Autowired
private ReleaseMessageRepository releaseMessageRepository;
/**
 * 从 DB 中扫描 ReleaseMessage 表的频率，单位：毫秒
 */
private int databaseScanInterval;
/**
 * 监听器数组
 */
private List<ReleaseMessageListener> listeners;
/**
 * 定时任务服务
 */
private ScheduledExecutorService executorService;
/**
 * 最后扫描到的 ReleaseMessage 的编号
 */
private long maxIdScanned;

public ReleaseMessageScanner() {
    // 创建监听器数组
    listeners = Lists.newCopyOnWriteArrayList();
    // 创建 ScheduledExecutorService 对象
    executorService = Executors.newScheduledThreadPool(1, ApolloThreadFactory.create("ReleaseMessageScanner", true));
}

3.2 初始化 Scan 任务

#afterPropertiesSet() 方法，通过 Spring 调用，初始化 Scan 任务。代码如下：

 1 @Override
 2 public void afterPropertiesSet() {
 3     // 从 ServerConfig 中获得频率
 4     databaseScanInterval = bizConfig.releaseMessageScanIntervalInMilli();
 5     // 获得最大的 ReleaseMessage 的编号
 6     maxIdScanned = loadLargestMessageId();
 7     // 创建从 DB 中扫描 ReleaseMessage 表的定时任务
 8     executorService.scheduleWithFixedDelay((Runnable) () -> {
 9         // 【TODO 6001】Tracer 日志
10         Transaction transaction = Tracer.newTransaction("Apollo.ReleaseMessageScanner", "scanMessage");
11         try {
12             // 从 DB 中，扫描 ReleaseMessage 们
13             scanMessages();
14             // 【TODO 6001】Tracer 日志
15             transaction.setStatus(Transaction.SUCCESS);
16         } catch (Throwable ex) {
17             // 【TODO 6001】Tracer 日志
18             transaction.setStatus(ex);
19             logger.error("Scan and send message failed", ex);
20         } finally {
21             // 【TODO 6001】Tracer 日志
22             transaction.complete();
23         }
24     }, databaseScanInterval, databaseScanInterval, TimeUnit.MILLISECONDS);
25 }

第 4 行：调用 BizConfig#releaseMessageScanIntervalInMilli() 方法，从 ServerConfig 中获得频率，单位：毫秒。可通过 "apollo.message-scan.interval" 配置，默认：1000 ms 。

第 6 行：调用 #loadLargestMessageId() 方法，获得最大的 ReleaseMessage 的编号。代码如下：

/**
 * find largest message id as the current start point
 *
 * @return current largest message id
 */
private long loadLargestMessageId() {
    ReleaseMessage releaseMessage = releaseMessageRepository.findTopByOrderByIdDesc();
    return releaseMessage == null ? 0 : releaseMessage.getId();
}

第 8 至 24 行：调用 ExecutorService#scheduleWithFixedDelay(Runnable) 方法，创建从 DB 中扫描 ReleaseMessage 表的定时任务。
- 第 13 行：调用 #scanMessages() 方法，从 DB 中，扫描新的 ReleaseMessage 们。

#scanMessages() 方法，循环扫描消息，直到没有新的 ReleaseMessage 为止。代码如下：

private void scanMessages() {
    boolean hasMoreMessages = true;
    while (hasMoreMessages && !Thread.currentThread().isInterrupted()) {
        hasMoreMessages = scanAndSendMessages();
    }
}

#scanAndSendMessages() 方法，扫描消息，并返回是否继续有新的 ReleaseMessage 可以继续扫描。代码如下：

 1 private boolean scanAndSendMessages() {
 2     // 获得大于 maxIdScanned 的 500 条 ReleaseMessage 记录，按照 id 升序
 3     // current batch is 500
 4     List<ReleaseMessage> releaseMessages = releaseMessageRepository.findFirst500ByIdGreaterThanOrderByIdAsc(maxIdScanned);
 5     if (CollectionUtils.isEmpty(releaseMessages)) {
 6         return false;
 7     }
 8     // 触发监听器
 9     fireMessageScanned(releaseMessages);
10     // 获得新的 maxIdScanned ，取最后一条记录
11     int messageScanned = releaseMessages.size();
12     maxIdScanned = releaseMessages.get(messageScanned - 1).getId();
13     // 若拉取不足 500 条，说明无新消息了
14     return messageScanned == 500;
15 }

第 4 至 7 行：调用 ReleaseMessageRepository#findFirst500ByIdGreaterThanOrderByIdAsc(maxIdScanned) 方法，获得大于 maxIdScanned 的 500 条 ReleaseMessage 记录，按照 id 升序。
第 9 行：调用 #fireMessageScanned(List<ReleaseMessage> messages) 方法，触发监听器们。
第 10 至 12 行：获得新的 maxIdScanned ，取最后一条记录。
第 14 行：若拉取不足 500 条，说明无新消息了。

3.3 fireMessageScanned

#fireMessageScanned(List<ReleaseMessage> messages) 方法，触发监听器，处理 ReleaseMessage 们。代码如下：

private void fireMessageScanned(List<ReleaseMessage> messages) {
    for (ReleaseMessage message : messages) { // 循环 ReleaseMessage
        for (ReleaseMessageListener listener : listeners) { // 循环 ReleaseMessageListener
            try {
                // 触发监听器
                listener.handleMessage(message, Topics.APOLLO_RELEASE_TOPIC);
            } catch (Throwable ex) {
                Tracer.logError(ex);
                logger.error("Failed to invoke message listener {}", listener.getClass(), ex);
            }
        }
    }
}

3.4 ReleaseMessageListener

com.ctrip.framework.apollo.biz.message.ReleaseMessageListener ，ReleaseMessage 监听器接口。代码如下：

public interface ReleaseMessageListener {

    /**
     * 处理 ReleaseMessage
     *
     * @param message
     * @param channel 通道（主题）
     */
    void handleMessage(ReleaseMessage message, String channel);

}

ReleaseMessageListener 实现子类如下图：

例如，NotificationControllerV2 得到配置发布的 AppId+Cluster+Namespace 后，会通知对应的客户端。? 具体的代码实现，我们后面会看。

那么 ReleaseMessageScanner 中的listeners 属性，监听器数组，是什么初始化进去的呢。通过 #addMessageListener(ReleaseMessageListener) 方法，注册 ReleaseMessageListener 。在 MessageScannerConfiguration 中，调用该方法，初始化 ReleaseMessageScanner 的监听器们。代码如下：

@Configuration
static class MessageScannerConfiguration {

    @Autowired
    private NotificationController notificationController;
    @Autowired
    private ConfigFileController configFileController;
    @Autowired
    private NotificationControllerV2 notificationControllerV2;
    @Autowired
    private GrayReleaseRulesHolder grayReleaseRulesHolder;
    @Autowired
    private ReleaseMessageServiceWithCache releaseMessageServiceWithCache;
    @Autowired
    private ConfigService configService;

    @Bean
    public ReleaseMessageScanner releaseMessageScanner() {
        ReleaseMessageScanner releaseMessageScanner = new ReleaseMessageScanner();
        // 0. handle release message cache
        releaseMessageScanner.addMessageListener(releaseMessageServiceWithCache);
        // 1. handle gray release rule
        releaseMessageScanner.addMessageListener(grayReleaseRulesHolder);
        // 2. handle server cache
        releaseMessageScanner.addMessageListener(configService);
        releaseMessageScanner.addMessageListener(configFileController);
        // 3. notify clients
        releaseMessageScanner.addMessageListener(notificationControllerV2);
        releaseMessageScanner.addMessageListener(notificationController);
        return releaseMessageScanner;
    }
}

4 小结

好啦，到这里发布配置的过程我们已经看了一半哈，这节我们主要看了发布配置后，消息的发送以及消费，消费里会通过遍历监听器进行处理，有理解不对的地方欢迎指正哈。