筆者是 RocketMQ 的忠實粉絲，在閱讀源碼的過程中，學(xué)習(xí)到了很多編程技巧。

這篇文章，筆者結(jié)合 RocketMQ 源碼，分享并發(fā)編程三大神器的相關(guān)知識點。

1 CountDownLatch 實現(xiàn)網(wǎng)絡(luò)同步請求

CountDownLatch 是一個同步工具類，用來協(xié)調(diào)多個線程之間的同步，它能夠使一個線程在等待另外一些線程完成各自工作之后，再繼續(xù)執(zhí)行。

下圖是 CountDownLatch 的核心方法：

我們可以認(rèn)為它內(nèi)置一個計數(shù)器，構(gòu)造函數(shù)初始化計數(shù)值。每當(dāng)線程執(zhí)行 countDown 方法，計數(shù)器的值就會減一，當(dāng)計數(shù)器的值為 0 時，表示所有的任務(wù)都執(zhí)行完成，然后在 CountDownLatch 上等待的線程就可以恢復(fù)執(zhí)行接下來的任務(wù)。

舉例，數(shù)據(jù)庫有100萬條數(shù)據(jù)需要處理，單線程執(zhí)行比較慢，我們可以將任務(wù)分為5個批次，線程池按照每個批次執(zhí)行，當(dāng)5個批次整體執(zhí)行完成后，打印出任務(wù)執(zhí)行的時間 。

 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("任務(wù)執(zhí)行耗時:" + (System.currentTimeMillis() - start) + "毫秒");

溫習(xí)完 CountDownLatch 的知識點，回到 RocketMQ 源碼。

筆者在沒有接觸網(wǎng)絡(luò)編程之前，一直很疑惑，網(wǎng)絡(luò)同步請求是如何實現(xiàn)的？

同步請求指：客戶端線程發(fā)起調(diào)用后，需要在指定的超時時間內(nèi)，等到響應(yīng)結(jié)果，才能完成本次調(diào)用。如果超時時間內(nèi)沒有得到結(jié)果，那么會拋出超時異常。

RocketMQ 的同步發(fā)送消息接口見下圖：

追蹤源碼，真正發(fā)送請求的方法是通訊模塊的同步請求方法 invokeSyncImpl 。

整體流程：

發(fā)送消息線程 Netty channel 對象調(diào)用 writeAndFlush 方法后 ，它的本質(zhì)是通過 Netty 的讀寫線程將數(shù)據(jù)包發(fā)送到內(nèi)核 , 這個過程本身就是異步的；

ResponseFuture 類中內(nèi)置一個 CountDownLatch 對象 ，responseFuture 對象調(diào)用 waitRepsone 方法，發(fā)送消息線程會阻塞 ；

客戶端收到響應(yīng)命令后， 執(zhí)行 processResponseCommand 方法，核心邏輯是執(zhí)行 ResponseFuture 的 putResponse 方法。

該方法的本質(zhì)就是填充響應(yīng)對象，并調(diào)用 countDownLatch 的 countDown 方法 , 這樣發(fā)送消息線程就不再阻塞。

CountDownLatch 實現(xiàn)網(wǎng)絡(luò)同步請求是非常實用的技巧，在很多開源中間件里，比如 Metaq ，Xmemcached 都有類似的實現(xiàn)。

2 ReadWriteLock 名字服務(wù)路由管理

讀寫鎖是一把鎖分為兩部分：讀鎖和寫鎖，其中讀鎖允許多個線程同時獲得，而寫鎖則是互斥鎖。

它的規(guī)則是：讀讀不互斥，讀寫互斥，寫寫互斥，適用于讀多寫少的業(yè)務(wù)場景。

我們一般都使用 ReentrantReadWriteLock ，該類實現(xiàn)了 ReadWriteLock 。ReadWriteLock 接口也很簡單，其內(nèi)部主要提供了兩個方法，分別返回讀鎖和寫鎖 。

 public interface ReadWriteLock {
    //獲取讀鎖
    Lock readLock();
    //獲取寫鎖
    Lock writeLock();
}

讀寫鎖的使用方式如下所示：

創(chuàng)建 ReentrantReadWriteLock 對象 , 當(dāng)使用 ReadWriteLock 的時候，并不是直接使用，而是獲得其內(nèi)部的讀鎖和寫鎖，然后分別調(diào)用 lock / unlock 方法 ;

private ReadWriteLock readWriteLock = new ReentrantReadWriteLock();

讀取共享數(shù)據(jù) ；

Lock readLock = readWriteLock.readLock();
readLock.lock();
try {
   // TODO 查詢共享數(shù)據(jù)
} finally {
   readLock.unlock();
}

寫入共享數(shù)據(jù)；

Lock writeLock = readWriteLock.writeLock();
writeLock.lock();
try {
   // TODO 修改共享數(shù)據(jù)
} finally {
   writeLock.unlock();
}

RocketMQ架構(gòu)上主要分為四部分，如下圖所示 :

Producer ：消息發(fā)布的角色，Producer 通過 MQ 的負(fù)載均衡模塊選擇相應(yīng)的 Broker 集群隊列進(jìn)行消息投遞，投遞的過程支持快速失敗并且低延遲。

Consumer ：消息消費的角色，支持以 push 推，pull 拉兩種模式對消息進(jìn)行消費。

BrokerServer ：Broker主要負(fù)責(zé)消息的存儲、投遞和查詢以及服務(wù)高可用保證。

NameServer ：名字服務(wù)是一個非常簡單的 Topic 路由注冊中心，其角色類似 Dubbo 中的zookeeper，支持Broker的動態(tài)注冊與發(fā)現(xiàn)。

NameServer 是一個幾乎無狀態(tài)節(jié)點，可集群部署，節(jié)點之間無任何信息同步。Broker 啟動之后會向所有 NameServer 定期（每 30s）發(fā)送心跳包(路由信息)，NameServer 會定期掃描 Broker 存活列表，如果超過 120s 沒有心跳則移除此 Broker 相關(guān)信息，代表下線。

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

路由信息通過幾個 HashMap 來保存，當(dāng) Broker 向 Nameserver 發(fā)送心跳包（路由信息），Nameserver 需要對 HashMap 進(jìn)行數(shù)據(jù)更新，但我們都知道 HashMap 并不是線程安全的，高并發(fā)場景下，容易出現(xiàn) CPU 100% 問題，所以更新 HashMap 時需要加鎖，RocketMQ 使用了  JDK 的讀寫鎖 ReentrantReadWriteLock 。

更新路由信息，操作寫鎖

查詢主題信息，操作讀鎖

讀寫鎖適用于讀多寫少的場景，比如名字服務(wù)，配置服務(wù)等。

3 CompletableFuture 異步消息處理

RocketMQ 主從架構(gòu)中，主節(jié)點與從節(jié)點之間數(shù)據(jù)同步/復(fù)制的方式有同步雙寫和異步復(fù)制兩種模式。

異步復(fù)制是指消息在主節(jié)點落盤成功后就告訴客戶端消息發(fā)送成功，無需等待消息從主節(jié)點復(fù)制到從節(jié)點，消息的復(fù)制由其他線程完成。

同步雙寫是指主節(jié)點將消息成功落盤后，需要等待從節(jié)點復(fù)制成功，再告訴客戶端消息發(fā)送成功。

同步雙寫模式是阻塞的，筆者按照 RocketMQ 4.6.1 源碼，整理出主節(jié)點處理一個發(fā)送消息的請求的時序圖。

整體流程：

生產(chǎn)者將消息發(fā)送到 Broker , Broker 接收到消息后，發(fā)送消息處理器 SendMessageProcessor 的執(zhí)行線程池SendMessageExecutor 線程池來處理發(fā)送消息命令；

執(zhí)行 ComitLog 的 putMessage 方法；

ComitLog 內(nèi)部先執(zhí)行 appendMessage 方法；

然后提交一個 GroupCommitRequest 到同步復(fù)制服務(wù) HAService  ,等待 HAService 通知 GroupCommitRequest 完成；

返回寫入結(jié)果并響應(yīng)客戶端 。

我們可以看到：發(fā)送消息的執(zhí)行線程需要等待消息復(fù)制從節(jié)點 , 并將消息返回給生產(chǎn)者才能開始處理下一個消息。

RocketMQ 4.6.1 源碼中，執(zhí)行線程池的線程數(shù)量是 1 ，假如線程處理主從同步速度慢了，系統(tǒng)在這一瞬間無法處理新的發(fā)送消息請求，造成 CPU 資源無法被充分利用 , 同時系統(tǒng)的吞吐量也會降低。

那么優(yōu)化同步雙寫呢 ？

從 RocketMQ 4.7 開始，RocketMQ 引入了 CompletableFuture 實現(xiàn)了異步消息處理 。

發(fā)送消息的執(zhí)行線程不再等待消息復(fù)制到從節(jié)點后再處理新的請求，而是提前生成 CompletableFuture 并返回 ;

HAService 中的線程在復(fù)制成功后，調(diào)用 CompletableFuture 的 complete 方法，通知 remoting 模塊響應(yīng)客戶端（線程池：PutMessageExecutor ） 。

我們分析下 RocketMQ 4.9.4 核心代碼：

Broker 接收到消息后，發(fā)送消息處理器 SendMessageProcessor 的執(zhí)行線程池SendMessageExecutor 線程池來處理發(fā)送消息命令；

調(diào)用 SendMessageProcessor 的 asyncProcessRequest 方法；

調(diào)用 Commitlog 的 aysncPutMessage 方法寫入消息 ；

這段代碼中，當(dāng) commitLog 執(zhí)行完 appendMessage 后， 需要執(zhí)行刷盤任務(wù)和同步復(fù)制兩個任務(wù)。但這兩個任務(wù)并不是同步執(zhí)行，而是異步的方式。

復(fù)制線程復(fù)制消息后，喚醒 future ；

組裝響應(yīng)命令 ，并將響應(yīng)命令返回給客戶端。

為了便于理解這一段消息發(fā)送處理過程的線程模型，筆者在 RocketMQ 源碼中做了幾處埋點，修改 Logback 的日志配置，發(fā)送一條普通的消息，觀察服務(wù)端日志。

從日志中，我們可以觀察到：

發(fā)送消息的執(zhí)行線程（圖中紅色）在執(zhí)行完創(chuàng)建刷盤 Future 和同步復(fù)制 future 之后，并沒有等待這兩個任務(wù)執(zhí)行完成，而是在結(jié)束 asyncProcessRequest 方法后就可以處理發(fā)送消息請求了 ；

刷盤線程和復(fù)制線程執(zhí)行完各自的任務(wù)后，喚醒 future，然后通過刷盤線程組裝存儲結(jié)果，最后通過 PutMessageExecutor 線程池（圖中黃色）將響應(yīng)命令返回給客戶端。

筆者一直認(rèn)為：異步是更細(xì)粒度的使用系統(tǒng)資源的一種方式，在異步消息處理的過程中，通過 CompletableFuture  這個神器，各個線程各司其職，優(yōu)雅且高效的提升了 RocketMQ 的性能。