品 RocketMQ 源码，学习并发编程三大神器

这篇文章，笔者结合 RocketMQ 源码，分享并发编程三大神器的相关知识点。

1 CountDownLatch 实现网络同步请求

CountDownLatch 是一个同步工具类，用来协调多个线程之间的同步，它能够使一个线程在等待另外一些线程完成各自工作之后，再继续执行。

下图是 CountDownLatch 的核心方法：

我们可以认为它内置一个计数器，构造函数初始化计数值。每当线程执行 countDown 方法，计数器的值就会减一，当计数器的值为 0 时，表示所有的任务都执行完成，然后在 CountDownLatch 上等待的线程就可以恢复执行接下来的任务。

举例，数据库有100万条数据需要处理，单线程执行比较慢，我们可以将任务分为5个批次，线程池按照每个批次执行，当5个批次整体执行完成后，打印出任务执行的时间 。

 long start = System.currentTimeMillis();
 ExecutorService executorService = Executors.newFixedThreadPool(10);
 int batchSize = 5;
 CountDownLatch countDownLatch = new CountDownLatch(batchSize);
 for (int i = 0; i < batchSize; i++) {
   final int batchNumber = i;
   executorService.execute(new Runnable() {
      @Override
      public void run(){
        try {
           doSomething(batchNumber);
        } catch (Exception e) {
           e.printStackTrace();
        } finally {
           countDownLatch.countDown();
        }
      }
   });
}
countDownLatch.await();
System.out.println("任务执行耗时:" + (System.currentTimeMillis() - start) + "毫秒");

温习完 CountDownLatch 的知识点，回到 RocketMQ 源码。

笔者在没有接触网络编程之前，一直很疑惑，网络同步请求是如何实现的？

同步请求指：客户端线程发起调用后，需要在指定的超时时间内，等到响应结果，才能完成本次调用。如果超时时间内没有得到结果，那么会抛出超时异常。

RocketMQ 的同步发送消息接口见下图：

追踪源码，真正发送请求的方法是通讯模块的同步请求方法 invokeSyncImpl 。

整体流程：

发送消息线程 Netty channel 对象调用 writeAndFlush 方法后 ，它的本质是通过 Netty 的读写线程将数据包发送到内核 , 这个过程本身就是异步的；

ResponseFuture 类中内置一个 CountDownLatch 对象 ，responseFuture 对象调用 waitRepsone 方法，发送消息线程会阻塞 ；

客户端收到响应命令后， 执行 processResponseCommand 方法，核心逻辑是执行 ResponseFuture 的 putResponse 方法。

该方法的本质就是填充响应对象，并调用 countDownLatch 的 countDown 方法 , 这样发送消息线程就不再阻塞。

CountDownLatch 实现网络同步请求是非常实用的技巧，在很多开源中间件里，比如 Metaq ，Xmemcached 都有类似的实现。

2 ReadWriteLock 名字服务路由管理

读写锁是一把锁分为两部分：读锁和写锁，其中读锁允许多个线程同时获得，而写锁则是互斥锁。

它的规则是：读读不互斥，读写互斥，写写互斥，适用于读多写少的业务场景。

我们一般都使用 ReentrantReadWriteLock ，该类实现了 ReadWriteLock 。ReadWriteLock 接口也很简单，其内部主要提供了两个方法，分别返回读锁和写锁 。

 public interface ReadWriteLock {
    //获取读锁
    Lock readLock();
    //获取写锁
    Lock writeLock();
}

读写锁的使用方式如下所示：

创建 ReentrantReadWriteLock 对象 , 当使用 ReadWriteLock 的时候，并不是直接使用，而是获得其内部的读锁和写锁，然后分别调用 lock / unlock 方法 ;

private ReadWriteLock readWriteLock = new ReentrantReadWriteLock();

读取共享数据 ；

Lock readLock = readWriteLock.readLock();
readLock.lock();
try {
   // TODO 查询共享数据
} finally {
   readLock.unlock();
}

写入共享数据；

Lock writeLock = readWriteLock.writeLock();
writeLock.lock();
try {
   // TODO 修改共享数据
} finally {
   writeLock.unlock();
}

RocketMQ架构上主要分为四部分，如下图所示 :

Producer ：消息发布的角色，Producer 通过 MQ 的负载均衡模块选择相应的 Broker 集群队列进行消息投递，投递的过程支持快速失败并且低延迟。

Consumer ：消息消费的角色，支持以 push 推，pull 拉两种模式对消息进行消费。

BrokerServer ：Broker主要负责消息的存储、投递和查询以及服务高可用保证。

NameServer ：名字服务是一个非常简单的 Topic 路由注册中心，其角色类似 Dubbo 中的zookeeper，支持Broker的动态注册与发现。

NameServer 是一个几乎无状态节点，可集群部署，节点之间无任何信息同步。Broker 启动之后会向所有 NameServer 定期（每 30s）发送心跳包(路由信息)，NameServer 会定期扫描 Broker 存活列表，如果超过 120s 没有心跳则移除此 Broker 相关信息，代表下线。

那么 NameServer 如何保存路由信息呢？

路由信息通过几个 HashMap 来保存，当 Broker 向 Nameserver 发送心跳包（路由信息），Nameserver 需要对 HashMap 进行数据更新，但我们都知道 HashMap 并不是线程安全的，高并发场景下，容易出现 CPU 100% 问题，所以更新 HashMap 时需要加锁，RocketMQ 使用了  JDK 的读写锁 ReentrantReadWriteLock 。

更新路由信息，操作写锁

查询主题信息，操作读锁

读写锁适用于读多写少的场景，比如名字服务，配置服务等。

3 CompletableFuture 异步消息处理

RocketMQ 主从架构中，主节点与从节点之间数据同步/复制的方式有同步双写和异步复制两种模式。

异步复制是指消息在主节点落盘成功后就告诉客户端消息发送成功，无需等待消息从主节点复制到从节点，消息的复制由其他线程完成。

同步双写是指主节点将消息成功落盘后，需要等待从节点复制成功，再告诉客户端消息发送成功。

同步双写模式是阻塞的，笔者按照 RocketMQ 4.6.1 源码，整理出主节点处理一个发送消息的请求的时序图。

整体流程：

生产者将消息发送到 Broker , Broker 接收到消息后，发送消息处理器 SendMessageProcessor 的执行线程池SendMessageExecutor 线程池来处理发送消息命令；

执行 ComitLog 的 putMessage 方法；

ComitLog 内部先执行 appendMessage 方法；

然后提交一个 GroupCommitRequest 到同步复制服务 HAService  ,等待 HAService 通知 GroupCommitRequest 完成；

返回写入结果并响应客户端 。

我们可以看到：发送消息的执行线程需要等待消息复制从节点 , 并将消息返回给生产者才能开始处理下一个消息。

RocketMQ 4.6.1 源码中，执行线程池的线程数量是 1 ，假如线程处理主从同步速度慢了，系统在这一瞬间无法处理新的发送消息请求，造成 CPU 资源无法被充分利用 , 同时系统的吞吐量也会降低。

那么优化同步双写呢 ？

从 RocketMQ 4.7 开始，RocketMQ 引入了 CompletableFuture 实现了异步消息处理 。

发送消息的执行线程不再等待消息复制到从节点后再处理新的请求，而是提前生成 CompletableFuture 并返回 ;

HAService 中的线程在复制成功后，调用 CompletableFuture 的 complete 方法，通知 remoting 模块响应客户端（线程池：PutMessageExecutor ） 。

我们分析下 RocketMQ 4.9.4 核心代码：

Broker 接收到消息后，发送消息处理器 SendMessageProcessor 的执行线程池SendMessageExecutor 线程池来处理发送消息命令；

调用 SendMessageProcessor 的 asyncProcessRequest 方法；

调用 Commitlog 的 aysncPutMessage 方法写入消息 ；

这段代码中，当 commitLog 执行完 appendMessage 后， 需要执行刷盘任务和同步复制两个任务。但这两个任务并不是同步执行，而是异步的方式。

复制线程复制消息后，唤醒 future ；

组装响应命令 ，并将响应命令返回给客户端。

为了便于理解这一段消息发送处理过程的线程模型，笔者在 RocketMQ 源码中做了几处埋点，修改 Logback 的日志配置，发送一条普通的消息，观察服务端日志。

从日志中，我们可以观察到：

发送消息的执行线程（图中红色）在执行完创建刷盘 Future 和同步复制 future 之后，并没有等待这两个任务执行完成，而是在结束 asyncProcessRequest 方法后就可以处理发送消息请求了 ；

刷盘线程和复制线程执行完各自的任务后，唤醒 future，然后通过刷盘线程组装存储结果，最后通过 PutMessageExecutor 线程池（图中黄色）将响应命令返回给客户端。

笔者一直认为：异步是更细粒度的使用系统资源的一种方式，在异步消息处理的过程中，通过 CompletableFuture  这个神器，各个线程各司其职，优雅且高效的提升了 RocketMQ 的性能。