RocketMQ 主從復制是 RocketMQ 高可用機制之一，數(shù)據(jù)可以從主節(jié)點復制到一個或多個從節(jié)點。

這篇文章，我們聊聊 RocketMQ 的主從復制，希望大家讀完之后，能夠理解主從復制的精髓。

圖片

一、同步與異步

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

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

圖片

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

1、同步復制

圖片

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

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

• 優(yōu)點：數(shù)據(jù)與服務都無單點故障，Master宕機情況下，消息無延遲，服務可用性與數(shù)據(jù)可用性都非常高；
• 缺點：性能比異步復制模式略低（大約低10%左右），發(fā)送單個消息的 RT 會略高，且目前版本在主節(jié)點宕機后，備機不能自動切換為主機。

2、異步復制

圖片

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

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

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

• 主從服務器同步主題，消費者進度，延遲消費進度，消費者配置數(shù)據(jù)
• 主從服務器同步消息數(shù)據(jù)

二、元數(shù)據(jù)復制

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

圖片

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

圖片

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

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

圖片

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

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

圖片

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

• 讀服務：處理 Slave 發(fā)送的請求
• 寫服務：用于向 Slave 傳輸數(shù)據(jù)

圖片

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

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

圖片

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

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

圖片

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

圖片

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

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

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

圖片

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

圖片

processReadEvent 方法的核心邏輯是設置 Slave 的當前進度 offset ，然后通知復制線程當前的復制進度。

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

圖片

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

圖片

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

圖片

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

圖片

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

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

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

圖片

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

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

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

圖片

最后 Broker 組裝響應命令 ，并將響應命令返回給客戶端。

五、總結

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

核心要點如下：

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

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

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

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

1. Master 啟動監(jiān)聽指定端口
2. Slave  啟動 HaClient 服務，和 Master 創(chuàng)建 TCP 鏈接
3. Slave 向 Master 上報存儲進度
4. Master 接收進度，消息文件中檢索該偏移量后的所有消息，并傳輸給 Slave
5. Slave 接收到數(shù)據(jù)后，將消息數(shù)據(jù) append 到本地的消息存儲。

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

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

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