RocketMQ 的高階功能
一、訊息儲存
分散式佇列高可用，資料要進行持久化儲存。
1、訊息儲存的流程
① 訊息生產者傳送訊息；
② MQ 收到訊息，將訊息持久化，在儲存中新增一條記錄；
③ 將ACK 返回給生成者；
④ MQ push 訊息給對應的消費者，然後等待消費者返回 ACK；
⑤ 如果訊息消費者在指定時間內返回 ACK，那麼 MQ 認為訊息被消費成功，在儲存中刪除訊息，即執行第 6 步；如果 MQ 在指定時間內沒有收到 ACK ，則認為訊息消費失敗，會嘗試重新 push，重複執行 第 ④ ⑤ ⑥ 步；
6、MQ 刪除訊息。
2、訊息儲存介質
檔案系統：
目前業界較為常用的幾款產品（RocketMQ/Kafka/RabbitMQ）均採用的是訊息刷盤至所部署虛擬機器 / 物理機的檔案系統來做持久化（刷盤一般可以分為非同步刷盤和同步刷盤兩種模式）。訊息刷盤為訊息儲存提供了一種高效率、高可靠性和高效能的資料持久化方式。除非部署 MQ 機器本身或是本地磁碟掛了，否則一般是不會出現無法持久化的故障問題。RocketMQ 的訊息用順序寫，保證了訊息儲存的速度。
3、訊息儲存結構
① CommitLog：儲存訊息的元資料。
② ConsumerQueue：儲存訊息在 CommitLog 的索引。
③ IndexFile：為了訊息查詢提供了一種透過 key 或時間區間來查詢訊息的方法，這種透過 IndexFile 來查詢訊息的方法不影響傳送與消費訊息的主流程。
4、刷盤機制
① 同步刷盤
在返回寫成功狀態時，訊息已經被寫入磁碟。
具體：訊息寫入記憶體的 pagecache，立刻通知刷盤執行緒刷盤，然後等待刷盤完成，刷盤執行緒執行完成後喚醒等待執行緒，返回訊息寫成功狀態。
② 非同步刷盤
在返回寫成功狀態時，訊息只是被寫入了記憶體的pagecache，寫操作返回快，吞吐量大，當記憶體的訊息積累到一定程度時，統一觸發寫磁碟動作，快速寫入。
③ 配置
透過 Broker 配置檔案 flushDiskType 引數設定的，引數配置為 SYNC_FLUSH 或者 ASYNC_FLUSH 中的一個。
二、高可用機制
叢集
RocketMQ 分散式叢集是透過 Broker 的 Master 和 Slave 配合。
在 Broker 配置中，引數 brokerId 為 0 表明 Broker 是 Master，大於 0 表明是 Slave，同時 brokerRole 引數說明 Broker 是 Master 和 Slave。
Master 角色的 Broker 支援讀和寫，Slave 角色的 Broker 僅支援讀取，也就是 Producer 只能和 Master 角色的 Broker 連線寫入訊息，Consumer 可以連線 Master 或 Slave 角色的 Broker 來讀取資訊。
（從上藍色圖中我們也可以看出，生產者叢集只和 Master 相連，沒有和 Slave 相連，而 消費者叢集是和 Master 和 Slave 都相連）
1、訊息傳送的高可用
① “重試”機制
在 client 端，傳送訊息的方式有：同步（SYNC）、非同步（ASYNC）、單向（ONEWAY）。
那麼可以知道，retryTimesWhenSendFailed 決定同步方法重試次數，預設重試次數為 3 次。
重試機制提高了訊息傳送的成功率。
② Broker 規避
透過 sendLatencyFaultEnable 的值選擇佇列的方式，預設 false ，不啟動 broker 故障延遲機制，當為 true 時 ，啟動 broker 故障延遲機制。
a。當 sendLatencyFaultEnable = false 時，輪詢選擇佇列，很可能下一次選擇的佇列還是在已經宕機的 broker，沒有辦法規避故障的 broker，因此訊息傳送很可能會再次失敗，重試傳送造成了不必要的效能損失。
b。 當 sendLatencyFaultEnable=true 時
預設機制一致，不同的是多了判斷是否可用，呼叫了 latencyFaultTolerance。isAvailable （mq。getBrokerName （）） 判斷。
2、訊息消費的高可用
在 consumer 的配置檔案中，不需要設定從 Master 或 Slave 讀，當 Master 不可用或繁忙時， consumer 會自動切換到 Slave，從而達到高可用。
3、訊息的主從複製
如果一個 Broker 組有 Master 和 Slave，訊息需要從 Master 複製到 Slave 上，有同步和非同步兩種複製方式。
同步複製是等 Master 和 Slave 均寫入成功後才反饋給客戶端寫成功狀態；
優點：如果 Master 故障， Slave 有全部備份資料，容易恢復；
缺點：同步複製增大資料寫入延遲，降低系統吞吐量。
非同步複製是隻要 Master 寫入成功即可反饋給客戶端寫成功狀態
優點：系統擁有較低延遲和較高吞吐量；
缺點：如果 Master 故障，有些資料 Slave 沒有寫入，可能造成資料丟失。
同步複製和非同步複製是透過 Broker 配置檔案裡的 brokerRole 引數進行設定的，這個引數可以被設定成 ASYNC_ MASTER、SYNC_ MASTER、SLAVE 三個值中的一個。
三、負載均衡
1、 Producer 負載均衡
預設透過輪詢選擇不同的佇列
2、Consumer 負載均衡
在 Consumer 中，我們可以設定消費模型是叢集模式（預設選擇）還是廣播模式
叢集模式下有 5 種演算法策略
① 平均分配策略 （預設）（AllocateMessageQueueAveragely）
② 環形分配策略 （AllocateMessageQueueAveragelyByCircle）
③ 手動配置分配策略 （AllocateMessageQueueByConfig）
④ 機房分配策略 （AllocateMessageQueueByMachineRoom）
⑤ 一致性雜湊分配策略 （AllocateMessageQueueConsistentHash）
四、“訊息重試”
1、順序訊息的重試
對於順序訊息，當消費者消費訊息失敗後，訊息佇列RocketMQ會自動不斷進行訊息重試（每次間隔時間為1秒） ，這時，應用會出現訊息消費被阻塞的情況。（重試需要保證順序，可能造成阻塞）
因此，在使用順序訊息時，務必保證應用能夠及時監控並處理消費失敗的情況，避免阻塞現象的發生。
2、無序訊息的重試
對於無序訊息(普通、定時
、延時、事務訊息），當消費者消費訊息失敗時，您可以透過設定返回狀態達到訊息重試的結果。
無序訊息的重試只針對叢集消費方式生效；廣播方式不提供失敗重試特性，即消費失敗後，失敗訊息不再重試，繼續消費新的。
最多重試 16 次
如果 16 次還沒消費，會進入“死信”佇列。
① 訊息失敗後，重試配置方式，在訊息監聽器介面中
a。返回ConsumeConcurrentlyStatus。 RECONSUME_LATER（推薦）
b。返回 null，訊息重試
c。直接丟擲異常，訊息重試
② 訊息失敗後，不重試配置方式，try。。。catch，返回ConsumeConcurrentlyStatus。CONSUME_SUCCESS
③ 自定義訊息最大重試次數
a。配置設定
b。透過獲取訊息重試次數，然後判斷
五、“死信佇列”（Dead-Letter Queue）
根據上文我們可以知道，當達到最大重試次數，若消費仍然失敗，表明消費者在正常情況下無法正確地消費該訊息，此時，訊息佇列不會立刻將訊息丟棄，而是將其傳送到該消費者對應的特殊佇列中。
1、“死信佇列”特性
① 不會被消費者正常消費
② 有效期與正常訊息相同，均為 3 天，3 天后自動刪除
③ 一個死信佇列對應一個 Group ID，而不是對應單個消費例項
④ 如果一個 Group ID 未產生死信佇列，訊息佇列 RocketMQ 不會為其建立相應的死信佇列
⑤ 一個死信佇列包含了對應 Group ID 產生的所有死信訊息，無論該訊息屬於哪個 Topic
2、檢視“死信佇列”
透過視覺化頁面 RocketMQ-console，點選 Topic，選擇 DLQ
從圖中，我們可以看出，死信佇列的 Topic 是 %DLQ% 開頭
3、死信佇列產生原因
① 訊息被拒絕
② 訊息 TTL 過期
③ 佇列達到最大長度（無法新增新資料）
4、死信佇列處理方案
① 丟棄
② 記錄死信入庫，然後做後續的業務分析和處理
③ 透過死信佇列，有負責監聽死信的應用程式進行處理
死信佇列對應 Topic 的許可權為 2，只有寫許可權，所以死信佇列沒有辦法讀取。
透過將死信佇列的許可權設定為讀寫許可權（6），然後直接透過訂閱對應的死信佇列即可。
注意：
6：同時支援讀寫 4：禁寫 2：禁讀
六、消費冪等
消費者在接受訊息以後，有必要根據業務的唯一 key 對訊息做冪等處理。
1、訊息重複的情況
① 傳送時訊息重複
② 投遞時訊息重複
③ 負載均衡時訊息重複
2、處理方式
因為 Message_ID 有可能出現衝突，冪等處理不建議用 Message_ID 作為依據，最好的方式是以業務唯一標識作冪等處理，業務唯一標識可以透過訊息 key 設定（在訊息類中設定）
歡迎關注 @Python 大星 ，一個會點 Python 的 Java 程式設計師。如果你有更好的想法，歡迎留言，一起探討，想說你就說啊！後面繼續分享 Java 的相關開發，人少就散了吧！@Python 大星
@Python 大星 | 文
