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

　　笔者是RocketMQ的忠实粉丝，在阅读源码的过程中，学习到了很多编程技巧。
　　这篇文章，笔者结合RocketMQ源码，分享并发编程三大神器的相关知识点。
　　1CountDownLatch实现网络同步请求
　　CountDownLatch是一个同步工具类，用来协调多个线程之间的同步，它能够使一个线程在等待另外一些线程完成各自工作之后，再继续执行。
　　下图是CountDownLatch的核心方法：
　　我们可以认为它内置一个计数器，构造函数初始化计数值。每当线程执行countDown方法，计数器的值就会减一，当计数器的值为0时，表示所有的任务都执行完成，然后在CountDownLatch上等待的线程就可以恢复执行接下来的任务。
　　举例，数据库有100万条数据需要处理，单线程执行比较慢，我们可以将任务分为5个批次，线程池按照每个批次执行，当5个批次整体执行完成后，打印出任务执行的时间。longstartSystem。currentTimeMillis（）；ExecutorServiceexecutorServiceExecutors。newFixedThreadPool（10）；intbatchSize5；CountDownLatchcountDownLatchnewCountDownLatch（batchSize）；for（inti0；ibatchSize；i）｛finalintbatchNumberi；executorService。execute（newRunnable（）｛Overridepublicvoidrun（）｛try｛doSomething（batchNumber）；｝catch（Exceptione）｛e。printStackTrace（）；｝finally｛countDownLatch。countDown（）；｝｝｝）；｝countDownLatch。await（）；System。out。println（任务执行耗时：（System。currentTimeMillis（）start）毫秒）；
　　温习完CountDownLatch的知识点，回到RocketMQ源码。
　　笔者在没有接触网络编程之前，一直很疑惑，网络同步请求是如何实现的？
　　同步请求指：客户端线程发起调用后，需要在指定的超时时间内，等到响应结果，才能完成本次调用。如果超时时间内没有得到结果，那么会抛出超时异常。
　　RocketMQ的同步发送消息接口见下图：
　　追踪源码，真正发送请求的方法是通讯模块的同步请求方法invokeSyncImpl。
　　整体流程：发送消息线程Nettychannel对象调用writeAndFlush方法后，它的本质是通过Netty的读写线程将数据包发送到内核，这个过程本身就是异步的；ResponseFuture类中内置一个CountDownLatch对象，responseFuture对象调用waitRepsone方法，发送消息线程会阻塞；
　　客户端收到响应命令后，执行processResponseCommand方法，核心逻辑是执行ResponseFuture的putResponse方法。
　　该方法的本质就是填充响应对象，并调用countDownLatch的countDown方法，这样发送消息线程就不再阻塞。
　　CountDownLatch实现网络同步请求是非常实用的技巧，在很多开源中间件里，比如Metaq，Xmemcached都有类似的实现。2ReadWriteLock名字服务路由管理
　　读写锁是一把锁分为两部分：读锁和写锁，其中读锁允许多个线程同时获得，而写锁则是互斥锁。
　　它的规则是：读读不互斥，读写互斥，写写互斥，适用于读多写少的业务场景。
　　我们一般都使用ReentrantReadWriteLock，该类实现了ReadWriteLock。ReadWriteLock接口也很简单，其内部主要提供了两个方法，分别返回读锁和写锁。publicinterfaceReadWriteLock｛获取读锁LockreadLock（）；获取写锁LockwriteLock（）；｝
　　读写锁的使用方式如下所示：创建ReentrantReadWriteLock对象，当使用ReadWriteLock的时候，并不是直接使用，而是获得其内部的读锁和写锁，然后分别调用lockunlock方法；privateReadWriteLockreadWriteLocknewReentrantReadWriteLock（）；读取共享数据；LockreadLockreadWriteLock。readLock（）；readLock。lock（）；try｛TODO查询共享数据｝finally｛readLock。unlock（）；｝写入共享数据；LockwriteLockreadWriteLock。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并不是线程安全的，高并发场景下，容易出现CPU100问题，所以更新HashMap时需要加锁，RocketMQ使用了JDK的读写锁ReentrantReadWriteLock。更新路由信息，操作写锁
　　查询主题信息，操作读锁
　　读写锁适用于读多写少的场景，比如名字服务，配置服务等。3CompletableFuture异步消息处理
　　RocketMQ主从架构中，主节点与从节点之间数据同步复制的方式有同步双写和异步复制两种模式。
　　异步复制是指消息在主节点落盘成功后就告诉客户端消息发送成功，无需等待消息从主节点复制到从节点，消息的复制由其他线程完成。
　　同步双写是指主节点将消息成功落盘后，需要等待从节点复制成功，再告诉客户端消息发送成功。
　　同步双写模式是阻塞的，笔者按照RocketMQ4。6。1源码，整理出主节点处理一个发送消息的请求的时序图。
　　整体流程：生产者将消息发送到Broker，Broker接收到消息后，发送消息处理器SendMessageProcessor的执行线程池SendMessageExecutor线程池来处理发送消息命令；执行ComitLog的putMessage方法；ComitLog内部先执行appendMessage方法；然后提交一个GroupCommitRequest到同步复制服务HAService，等待HAService通知GroupCommitRequest完成；返回写入结果并响应客户端。
　　我们可以看到：发送消息的执行线程需要等待消息复制从节点，并将消息返回给生产者才能开始处理下一个消息。
　　RocketMQ4。6。1源码中，执行线程池的线程数量是1，假如线程处理主从同步速度慢了，系统在这一瞬间无法处理新的发送消息请求，造成CPU资源无法被充分利用，同时系统的吞吐量也会降低。
　　那么优化同步双写呢？
　　从RocketMQ4。7开始，RocketMQ引入了CompletableFuture实现了异步消息处理。发送消息的执行线程不再等待消息复制到从节点后再处理新的请求，而是提前生成CompletableFuture并返回；HAService中的线程在复制成功后，调用CompletableFuture的complete方法，通知remoting模块响应客户端（线程池：PutMessageExecutor）。
　　我们分析下RocketMQ4。9。4核心代码：Broker接收到消息后，发送消息处理器SendMessageProcessor的执行线程池SendMessageExecutor线程池来处理发送消息命令；调用SendMessageProcessor的asyncProcessRequest方法；
　　调用Commitlog的aysncPutMessage方法写入消息；这段代码中，当commitLog执行完appendMessage后，需要执行刷盘任务和同步复制两个任务。但这两个任务并不是同步执行，而是异步的方式。复制线程复制消息后，唤醒future；组装响应命令，并将响应命令返回给客户端。
　　为了便于理解这一段消息发送处理过程的线程模型，笔者在RocketMQ源码中做了几处埋点，修改Logback的日志配置，发送一条普通的消息，观察服务端日志。
　　从日志中，我们可以观察到：发送消息的执行线程（图中红色）在执行完创建刷盘Future和同步复制future之后，并没有等待这两个任务执行完成，而是在结束asyncProcessRequest方法后就可以处理发送消息请求了；刷盘线程和复制线程执行完各自的任务后，唤醒future，然后通过刷盘线程组装存储结果，最后通过PutMessageExecutor线程池（图中黄色）将响应命令返回给客户端。
　　笔者一直认为：异步是更细粒度的使用系统资源的一种方式，在异步消息处理的过程中，通过CompletableFuture这个神器，各个线程各司其职，优雅且高效的提升了RocketMQ的性能。
　　如果我的文章对你有所帮助，还请帮忙点赞、在看、转发一下，你的支持会激励我输出更高质量的文章，非常感谢！