![]()
笔者是 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 源码。
笔者在没有接触网络编程之前，一直很疑惑，<strong style="font-size: inherit;line-height: inherit;color: rgb(255, 104, 39);">网络同步请求是如何实现的？</strong>
同步请求指：客户端线程发起调用后，需要在指定的超时时间内，等到响应结果，才能完成本次调用。如果超时时间内没有得到结果，那么会抛出超时异常。
RocketMQ 的同步发送消息接口见下图：
![]()
追踪源码，真正发送请求的方法是通讯模块的同步请求方法 invokeSyncImpl 。
![]()
整体流程：
- 发送消息线程 Netty channel 对象调用 writeAndFlush 方法后 ，它的本质是通过 Netty 的读写线程将数据包发送到内核 , 这个过程本身就是异步的；
- ResponseFuture 类中内置一个 CountDownLatch 对象 ，responseFuture 对象调用 waitRepsone 方法，发送消息线程会阻塞 ；
![]()
- 客户端收到响应命令后， 执行 processResponseCommand 方法，核心逻辑是执行 ResponseFuture 的 putResponse 方法。
<img src="https://oscimg.oschina.net/oscnet/up-3c7698bdd308393056f6aa00fc0d9ee38fb.png" style="zoom:115%;" />
该方法的本质就是填充响应对象，并调用 countDownLatch 的 countDown 方法 , 这样发送消息线程就不再阻塞。
CountDownLatch 实现网络同步请求是非常实用的技巧，在很多开源中间件里，比如 Metaq ，Xmemcached 都有类似的实现。
2 ReadWriteLock 名字服务路由管理
读写锁是一把锁分为两部分：读锁和写锁，其中读锁允许多个线程同时获得，而写锁则是互斥锁。
它的规则是：<strong style="font-size: inherit;line-height: inherit;color: rgb(255, 104, 39);">读读不互斥，读写互斥，写写互斥</strong>，适用于读多写少的业务场景。
我们一般都使用 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）发送心跳包(<strong style="font-size: inherit;line-height: inherit;color: rgb(255, 104, 39);">路由信息</strong>)，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 完成；
返回写入结果并响应客户端 。
我们可以看到：<strong style="font-size: inherit;line-height: inherit;color: rgb(255, 104, 39);">发送消息的执行线程需要等待消息复制从节点 , 并将消息返回给生产者才能开始处理下一个消息</strong>。
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 的性能。
如果我的文章对你有所帮助，还请帮忙点赞、在看、转发一下，你的支持会激励我输出更高质量的文章，非常感谢！
![]()
