1 同步與異步

在 RocketMQ 的集群模式中，Broker 分為 Master 與 Slave，一個 Master 可以對應(yīng)多個 Slave，但是一個 Slave 只能對應(yīng)一個 Master。

每個 Broker 與 Name Server 集群中的所有節(jié)點建立長連接，定時注冊 Topic 信息到所有 Name Server。

Master 節(jié)點負(fù)責(zé)接收客戶端的寫入請求，并將消息持久化到磁盤上。而 Slave 節(jié)點則負(fù)責(zé)從 Master 節(jié)點復(fù)制消息數(shù)據(jù)，并保持與 Master 節(jié)點的同步。

1、同步復(fù)制

每個 Master 配置一個 Slave ，有多對 Master-Slave ，HA 采用同步雙寫方式，即只有主備都寫成功，才向應(yīng)用返回成功。

這種模式的優(yōu)缺點如下：

優(yōu)點：數(shù)據(jù)與服務(wù)都無單點故障，Master宕機(jī)情況下，消息無延遲，服務(wù)可用性與數(shù)據(jù)可用性都非常高；

缺點：性能比異步復(fù)制模式略低（大約低10%左右），發(fā)送單個消息的 RT 會略高，且目前版本在主節(jié)點宕機(jī)后，備機(jī)不能自動切換為主機(jī)。

2、異步復(fù)制

每個 Master 配置一個 Slave ，有多對 Master-Slave ，HA 采用異步復(fù)制方式，主備有短暫消息延遲（毫秒級），這種模式的優(yōu)缺點如下：

優(yōu)點：即使磁盤損壞，消息丟失的非常少，且消息實時性不會受影響，同時Master宕機(jī)后，消費(fèi)者仍然可以從Slave消費(fèi)，而且此過程對應(yīng)用透明，不需要人工干預(yù)，性能同多 Master 模式幾乎一樣；

缺點：Master 宕機(jī)，磁盤損壞情況下會丟失少量消息 。

復(fù)制流程分為兩個部分：元數(shù)據(jù)復(fù)制和消息數(shù)據(jù)復(fù)制。

主從服務(wù)器同步主題，消費(fèi)者進(jìn)度，延遲消費(fèi)進(jìn)度，消費(fèi)者配置數(shù)據(jù)

主從服務(wù)器同步消息數(shù)據(jù)

2 元數(shù)據(jù)復(fù)制

Slave Broker 定時任務(wù)每隔 10 秒會同步元數(shù)據(jù)，包括主題，消費(fèi)進(jìn)度，延遲消費(fèi)進(jìn)度，消費(fèi)者配置。

同步主題時, Slave Broker 向 Master Broker 發(fā)送 RPC 請求，返回數(shù)據(jù)后，首先加入本地緩存里，然后持久化到本地。

3 消息數(shù)據(jù)復(fù)制

下圖是 Master 和 Slave 消息數(shù)據(jù)同步的流程圖。

1、Master 啟動后監(jiān)聽指定端口；

Master 啟動后創(chuàng)建 AcceptSocketService 服務(wù) , 用來創(chuàng)建客戶端到服務(wù)端的 TCP 鏈接。

RocketMQ 抽象了鏈接對象 HAConnection , HAConnection 會啟動兩個線程，分別用于讀服務(wù)和寫服務(wù)：

讀服務(wù)：處理 Slave 發(fā)送的請求

寫服務(wù)：用于向 Slave 傳輸數(shù)據(jù)

2、Slave 啟動后，嘗試連接 Master ，建立 TCP 連接；

HAClient 是客戶端 Slave 的核心類 ，負(fù)責(zé)和 Master 創(chuàng)建連接和數(shù)據(jù)交互。

客戶端在啟動后，首先嘗試連接 Master , 查詢當(dāng)前消息存儲中最大的物理偏移量 ，并存儲在變量 currentReportedOffset 里。

3、Slave 向 Master 匯報拉取消息偏移量；

上報進(jìn)度的數(shù)據(jù)格式是一個 Long 類型的 Offset , 8個字節(jié) , 非常簡潔 。

發(fā)送到 Socket 緩沖區(qū)后 , 修改最后一次的寫時間 lastWriteTimestamp 。

4、Master 解析請求偏移量，從消息文件中檢索該偏移量后的所有消息；

當(dāng) Slave 上報數(shù)據(jù)到 Master 時，觸發(fā) SelectionKey.OP_READ 事件，Master 將請求交由 ReadSocketService 服務(wù)處理：

當(dāng) Slave Broker 傳遞了自身 commitlog 的 maxPhyOffset 時，Master 會馬上中斷 selector.select(1000) ，執(zhí)行 processReadEvent 方法。

processReadEvent 方法的核心邏輯是設(shè)置 Slave 的當(dāng)前進(jìn)度 offset ，然后通知復(fù)制線程當(dāng)前的復(fù)制進(jìn)度。

寫服務(wù) WriteSocketService 從消息文件中檢索該偏移量后的所有消息（傳輸批次數(shù)據(jù)大小限制），并將消息數(shù)據(jù)發(fā)送給 Slave。

5、Slave 接收到數(shù)據(jù)，將消息數(shù)據(jù) append 到消息文件 commitlog 里 。

首先 HAClient 類中調(diào)用 dispatchReadRequest 方法 ， 解析出消息數(shù)據(jù) ；

然后將消息數(shù)據(jù) append 到本地的消息存儲。

4 同步的實現(xiàn)

從數(shù)據(jù)復(fù)制流程圖，我們發(fā)覺數(shù)據(jù)復(fù)制本身就是一個異步執(zhí)行的，但是同步是如何實現(xiàn)的呢？

Master Broker 接收到寫入消息的請求后 ，調(diào)用 Commitlog 的 aysncPutMessage 方法寫入消息。

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

但這兩個任務(wù)并不是同步執(zhí)行，而是異步的方式，使用了 CompletableFuture 這個異步神器。

當(dāng) HAConnection 讀服務(wù)接收到 Slave 的進(jìn)度反饋，發(fā)現(xiàn)消息數(shù)據(jù)復(fù)制成功，則喚醒 future 。

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

5 總結(jié)

RocketMQ 主從復(fù)制的實現(xiàn)思路非常簡潔，Slave 啟動一個線程，不斷從 Master 拉取 Commit Log 中的數(shù)據(jù)，然后在異步 build 出 Consume Queue 數(shù)據(jù)結(jié)構(gòu)。

核心要點如下：

1、主從復(fù)制包含元數(shù)據(jù)復(fù)制和消息數(shù)據(jù)復(fù)制兩個部分；

2、元數(shù)據(jù)復(fù)制

Slave Broker 定時任務(wù)每隔 10 秒向 Master Broker 發(fā)送 RPC 請求，將元數(shù)據(jù)同步到緩存后，然后持久化到磁盤里；

3、消息數(shù)據(jù)復(fù)制

Master 啟動監(jiān)聽指定端口

Slave 啟動 HaClient 服務(wù)，和 Master 創(chuàng)建 TCP 鏈接

Slave 向 Master 上報存儲進(jìn)度

Master 接收進(jìn)度，消息文件中檢索該偏移量后的所有消息，并傳輸給 Slave

Slave 接收到數(shù)據(jù)后，將消息數(shù)據(jù) append 到本地的消息存儲。

4、同步的實現(xiàn)

當(dāng) commitLog 執(zhí)行完 appendMessage 后， 需要執(zhí)行刷盤任務(wù)和同步復(fù)制兩個任務(wù)，這里用到了 CompletableFuture 這個異步神器。

當(dāng) HAConnection 讀服務(wù)接收到 Slave 的進(jìn)度反饋，發(fā)現(xiàn)消息數(shù)據(jù)復(fù)制成功，則喚醒 future 。最后 Broker 組裝響應(yīng)命令 ，并將響應(yīng)命令返回給客戶端 。

審核編輯：劉清