RocketMQ消息中間件實戰派（上下冊）

第1 篇 基礎
第1 章 初識RocketMQ ·························.2
1.1 認識分佈式架構 ····························.2
1.1.1 什麽是分佈式 ·······················.2
1.1.2 什麽是架構 ··························.3
1.1.3 分佈式架構的冰與火 ··············.5
1.2 認識RocketMQ ·····························.7
1.2.1 什麽是RocketMQ ··················.7
1.2.2 RocketMQ 與分佈式架構 ·······.12
1.3 搭建RocketMQ 環境 ····················.13
1.3.1 瞭解RocketMQ 相關的安裝包 .13
1.3.2 搭建單Master 的單機環境 ·····.15
1.3.3 搭建多Master 的集群環境 ·····.15
1.3.4 搭建單Master 和單Slave 的集群環境 ·································.16
1.3.5 【實例】搭建Raft 集群環境 ··.18
1.4 RocketMQ 5.0 的新特性 ················.20
第2 章 實現通信渠道 ··························.22
2.1 認識通信渠道 ·····························.22
2.2 通信渠道的架構 ··························.25
2.2.1 認識Netty ·························.25
2.2.2 RocketMQ 通信渠道的架構 ····.29
2.3 使用Netty 實現通信渠道的原理 ······.33
2.3.1 實現通信渠道的通信協議 ······.33
2.3.2 實現客戶端通信渠道 ············.38
2.3.3 實現服務端通信渠道 ············.41
2.4 使用客戶端連接服務端 ·················.47
2.4.1 分析模塊通信渠道的類型 ······.48
2.4.2 連接服務端 ························.50
2.5 【實例】修改通信渠道客戶端和服務端的線程模型 ·······················.54
第3 章 消息路由信息的無狀態存儲和管理 ······································.61
3.1 瞭解消息路由信息 ·······················.61
3.2 Name Server 的架構 ·····················.68
3.2.1 認識無狀態架構 ··················.68
3.2.2 認識Name Server 架構 ·········.69
3.2.3 認識地址服務架構 ···············.72
3.2.4 【實例】用地址服務動態更新客戶端中的Name Server 節點的地址信息 ··························.74
3.3 “使用Name Server 存儲和管理消息路由信息”的原理 ·······················.77
3.3.1 Name Server 的啟動流程 ·······.77
3.3.2 註冊Broker Server ···············.84
3.3.3 取消註冊的業務場景和取消註冊Broker Server ···············.89
3.3.4 存儲和管理消息路由信息 ······.93
3.4 【實例】啟動多個Name Server 節點，模擬故障以驗證Name Server 節點的無狀態性 ·································.101
第4 章 生產消息和消費消息 ···············.104
4.1 生產和消費消息的模式 ················.104
4.1.1 生產消息 ··························.104
4.1.2 消費消息 ··························.110
4.2 消費消息的類型 ·························.116
4.2.1 集群消息 ··························.116
4.2.2 廣播消息 ··························.118
4.2.3 順序消息 ··························.119
4.2.4 並發消息 ··························.125
4.2.5 延遲消息 ··························.126
4.2.6 事務消息 ··························.130
4.3 生產消息 ··································.134
4.3.1 生產者SDK 的3 種使用方式 ································.135
4.3.2 採用“同步”模式生產消息 ···.137
4.3.3 採用“異步”模式生產消息 ···.142
4.3.4 採用“最多發送一次”模式生產消息 ··························.148
4.4 消費消息 ··································.152
4.4.1 採用pull 模式消費消息 ········.153
4.4.2 採用push 模式消費消息 ·······.171
4.4.3 採用pop 模式消費消息 ········.183
4.5 採用“請求/應答”消息實現同步調用 ········································.197
4.5.1 “請求/應答”消息的架構 ····.197
4.5.2 分析“請求/應答”消息的過程 ································.198
4.6 【實例】生產者發送消息，消費者順序地消費消息 ·························.212
4.6.1 驗證“普通順序消息” ········.212
4.6.2 驗證“嚴格順序消息” ········.217
第5 章 存儲消息 ······························.219
5.1 認識存儲引擎 ····························.219
5.1.1 什麽是存儲引擎 ·················.219
5.1.2 存儲引擎的架構 ·················.221
5.2 認識存儲模型 ····························.222
5.2.1 消息模型 ··························.222
5.2.2 文件模型 ··························.229
5.2.3 【實例】利用源碼遠程調試存儲模型 ································.236
5.3 啟動存儲引擎 ····························.238
5.3.1 初始化Broker Server 和存儲引擎 ································.238
5.3.2 啟動Broker Server 和存儲引擎 ································.247
5.3.3 【實例】動態修改存儲引擎的配置參數 ··························.251
5.4 使用存儲引擎處理“儲存消息”的請求 ········································.253
5.4.1 接收並處理生產消息的請求 ································.254
5.4.2 存儲消息 ··························.257
5.4.3 【實例】利用存儲引擎實現批量地存儲消息 ·······················.262
5.5 對比存儲單條消息和批量消息的性能 ········································.263

第6 章 治理消息 ······························.268
6.1 如何治理消息 ····························.268
6.1.1 治理消息的目標 ·················.268
6.1.2 使用消息度量提供治理消息的依據 ································.276
6.1.3 使用消息管控實現治理消息的落地 ································.280
6.2 認識命令控制台和UI 控制台 ········.285
6.2.1 什麽是命令控制台和UI控制台 ·····························.285
6.2.2 啟動命令控制台和UI控制台 ·····························.289
6.2.3 使用對象池管理RocketMQDashboard 中通信渠道客戶端的核心類 ·····························.292
6.3 使用命令控制台治理消息 ·············.298
6.3.1 執行治理消息的命令 ···········.299
6.3.2 【實例】使用命令控制台，完成
RocketMQ 集群的擴容 ·········.322
6.3.3 【實例】使用命令控制台，完成
RocketMQ 集群的縮容 ·········.325
6.3.4 【實例】使用命令控制台，動態增加Topic 的讀/寫消息隊列的數量 ·················.332
6.4 使用UI 控制台治理消息 ··············.334
6.4.1 使用UI 控制台治理消息的原理 ································.335
6.4.2 【實例】使用UI 控制台手動地禁用消費者組 ····················.341
6.4.3 【實例】使用UI 控制台重置消費者進度 ·······················.343
第7 章 實現分佈式事務 ·····················.345
7.1 什麽是分佈式事務 ······················.345
7.1.1 本地事務 ··························.345
7.1.2 分佈式事務 ·······················.350
7.1.3 事務消息與數據的最終一致性 ·····························.354
7.2 使用“兩階段提交”模式處理事務消息 ········································.359
7.3 使用“定時回調”機制查詢本地事務狀態 ········································.370
7.4 【實例】架構師如何在電商項目中落地分佈式事務 ·························.380
7.4.1 分析業務場景 ····················.380
7.4.2 分析業務復雜度 ·················.381
7.4.3 分析技術復雜度 ·················.385
7.4.4 落地代碼 ··························.391
7.4.5 驗證接入的結果 ·················.392
7.5 【實例】使用泛娛樂業務驗證事務消息的故障轉移機制 ···················.393
7.5.1 準備環境 ··························.394
7.5.2 架構設計 ··························.394
7.5.3 代碼設計 ··························.395
7.5.4 驗證故障轉移 ····················.397
第8 章 實現消息的可追蹤性 ···············.402
8.1 消息的可追蹤性 ·························.402
8.1.1 認識消息的可追蹤性 ···········.402
8.1.2 為什麽需要消息的可追蹤性 ···.404
8.1.3 RocketMQ 支持的追蹤類型 ···.404
8.2 使用消息軌跡實現消息的可追蹤性 ··································.405
8.2.1 認識消息軌跡 ····················.405
8.2.2 軌跡樹的原理 ····················.412
8.2.3 【實例】開啟消息軌跡，追蹤生產消息、消費消息和結束事務消息的過程 ································.421
8.3 使用Hook 機制收集Trace 消息 ······.422
8.3.1 什麽是Hook 機制 ···············.422
8.3.2 “使用Hook 機制的實現類收集Trace 消息”的原理 ······.425
8.4 使用“定時線程任務”異步存儲Trace信息 ········································.428
8.4.1 為什麽要異步存儲 ··············.428
8.4.2 異步存儲Trace 信息的過程 ···.431
8.4.3 【實例】斷點調試“客戶端異步存儲Trace 信息”···············.441
8.5 使用“OpenTracing + Jaeger”實現消息的分佈式鏈路追蹤 ·························.444
8.5.1 認識OpenTracing ···············.444
8.5.2 認識Jaeger ·······················.446
8.5.3 “使用Jaeger 收集消息鏈路數據”的原理 ·····························.446
8.5.4 “使用OpenTracing 規範消息鏈路追蹤”的原理 ····················.448
8.5.5 【實例】收集普通消息和事務消息，使用Jaeger 實現可視化 ········.452
8.6 【實例】自定義Hook 和Trace 信息的派發類 ·····································.456
8.6.1 設計 ································.457
8.6.2 關鍵代碼分析 ····················.457
第9 章 保證消息的穩定性 ··················.461
9.1 什麽是消息的穩定性 ···················.461
9.1.1 認識穩定性 ·······················.461
9.1.2 認識消息的穩定性 ··············.466

9.1.3 消息穩定性技術 ·················.467
9.2 實現RocketMQ 的實時監控 ··········.472
9.2.1 認識RocketMQ Exporter 及其相關組件 ··························.472
9.2.2 實時監控的邏輯架構 ···········.474
9.2.3 “使用RocketMQ Exporter 實現實時監控”的原理 ··············.477
9.2.4 搭建監控環境 ····················.486
9.2.5 【實例】基於Prometheus 和Alert Manager 的RocketMQ實時告警 ··························.494
9.3 實現RocketMQ 客戶端的實時度量 ··································.503
9.3.1 認識Skywalking ·················.503
9.3.2 “使用Skywalking 實時度量RocketMQ 客戶端”的原理 ···.504
9.3.3 搭建監控環境 ····················.509
9.3.4 【實例】修改源碼動態設置一些延遲時間，觸發基於Skywalking和釘釘的告警閾值 ··············.511
9.4 【實例】使用RocketMQ Dashboard實時監控RocketMQ ····················.515
9.4.1 設計 ································.515
9.4.2 修改RocketMQ Dashboard 的源碼 ································.516
9.4.3 配置監控規則 ····················.517
9.4.4 初始化生產者和消費者服務 ··.517
9.4.5 驗證監控告警 ····················.518
9.5 【實例】改造RocketMQ 以支持Nacos 集群管理 ··························.520
9.5.1 架構設計 ··························.520
9.5.2 落地代碼 ··························.521
9.5.3 驗證 ································.524
第4 篇 高並發、高可用和高性能
第10 章 消息的高並發 ······················.528
10.1 消息的高並發概述 ·····················.528
10.1.1 什麽是高並發 ··················.528
10.1.2 為什麽要實現消息的高並發 ·····························.529
10.2 使用“管理通信渠道”實現消息的高並發 ·····································.530
10.2.1 通信渠道與高並發 ············.530
10.2.2 管理生產者與Broker Server之間的通信渠道 ·················.532
10.2.3 管理消費者與Broker Server之間的通信渠道 ·················.533
10.3 使用“提高消費者處理消息的功能”實現消息的高並發 ······················.534
10.3.1 消費者處理消息的功能與高並發 ·····························.534
10.3.2 使用“擴容消費者客戶端”來提高消費消息的並行性 ········.534
10.3.3 採用pop 模式來提高消費消息的速度 ································.535

10.3.4 【實例】實現“單進程單消費者的並發和批量消費消息” ·····························.536
10.3.5 【實例】實現“單進程多消費者實例並發生產和消費消息” ·····························.541
10.4 使用線程池實現消息的高並發 ······.550
10.4.1 為什麽要使用線程池 ·········.550
10.4.2 使用線程池實現消息的高並發的原理 ································.551
10.4.3 【實例】使用RocketMQ 自帶的線程實現生產和消費消息的並發性 ·····························.558
10.4.4 【實例】動態修改Consumer客戶端的核心線程數 ···········.559
10.5 採用“異步”模式實現消息的高並發 ····································.562
10.5.1 使用“回調+異步響應對象”實現生產者的“異步”模式 ································.562
10.5.2 使用“異步命令事件處理器”實現Broker Server 的“異步”模式 ································.563
10.5.3 使用“流水線”技術實現存儲引擎的“異步”模式 ···········.564
10.6 使用“負載均衡”實現消息的高並發 ·····································.566
10.6.1 什麽是負載均衡 ···············.566
10.6.2 “使用選擇器實現生產消息的負載均衡”的原理 ··············.576
10.6.3 “使用客戶端負載均衡實現消費消息的均衡性”的原理 ········.577
10.6.4 “使用服務端負載均衡實現消費消息的均衡性”的原理 ········.578
10.6.5 【實例】動態調整消費者客戶端的負載均衡策略 ·················.579
第11 章 消息的高可用 ·······················.583
11.1 什麽是消息的高可用 ··················.583
11.1.1 生產消息的高可用 ············.584
11.1.2 存儲消息的高可用 ············.585
11.1.3 消費消息的高可用 ············.585
11.2 實現生產消息的高可用——利用故障容錯機制 ··································.587
11.2.1 什麽是故障容錯機制 ·········.587
11.2.2 重試生產消息失敗的請求 ···.588
11.2.3 過濾故障的Broker Server ····.590
11.2.4 故障延遲 ························.591
11.2.5 【實例】在Spring Cloud Alibaba項目中，植入故障並驗證故障
容錯機制 ··························.594
11.3 實現消費消息的高可用——利用重試隊列和死信隊列 ·························.596
11.3.1 利用“重試隊列”實現高可用的核心原理 ··························.597
11.3.2 利用“死信隊列”實現高可用的核心原理 ··························.598
11.3.3 【實例】動態調整重試策略，並驗證消費失敗的高可用性 ···.599
11.3.4 【實例】在消息重試失敗後，消費死信隊列中的消息 ········.601

11.4 實現Broker Server 的高可用——利用主/從同步 ·································.602
11.4.1 什麽是主/從同步 ···············.602
11.4.2 實現主/從節點之間的通信渠道 ································.605
11.4.3 實現主/從節點之間消息的同步 ································.610
11.4.4 【實例】模擬Broker Server主/從同步的業務場景 ··········.614
11.5 實現Broker Server 的高可用——利用多副本架構（Dledger） ···············.617
11.5.1 什麽是多副本架構（DLedger） ·······················.617
11.5.2 “多副本架構的通信渠道”的原理 ································.618
11.5.3 “多副本架構的協議”的原理 ································.621
11.5.4 “多副本架構的手動選舉”的原理 ································.627
11.5.5 “多副本架構的自動選舉”的原理 ································.631
11.5.6 “多副本架構的消息存儲”的原理 ································.634
11.5.7 “多副本架構中集群之間消息同步”的原理 ····················.638
11.5.8 【實例】在主節點出現故障後，多副本架構（DLedger）自動選舉，並保持主/從消息一致性 ·······.645
11.6 【實例】模擬業務場景“從主/從架構升級為多副本架構”，並驗證消息的高可用性 ··································.651
11.7 【實例】使用微服務來驗證多副本的手動選舉功能的高可用性 ·············.654
第12 章 消息的高性能 ······················.657
12.1 什麽是消息的高性能 ·················.657
12.1.1 生產消息的高性能 ············.657
12.1.2 存儲消息的高性能 ············.659
12.1.3 消費消息的高性能 ············.660
12.2 使用本地緩存實現高性能 ···········.661
12.2.1 緩存客戶端通信渠道 ·········.661
12.2.2 緩存消息路由信息 ············.663
12.2.3 【實例】使用IDEA 調試“緩存客戶端通信渠道” ······.665
12.3 提升存儲消息的性能——利用CommitLog 文件順序寫 ···············.670
12.3.1 什麽是CommitLog 文件 ·····.670
12.3.2 CommitLog 文件的存儲格式 ································.672
12.3.3 利用“順序寫”存儲消息的原理 ································.673
12.4 提高消費消息的速度——利用ConsumeQueue 文件實現 ··············.676
12.4.1 什麽是ConsumeQueue文件 ································.677
12.4.2 構建ConsumeQueue 文件 ····.678
12.4.3 利用ConsumeQueue 文件提高消費消息的性能 ·················.683
12.5 提高查詢消息的速度——利用索引文件實現 ··································.684
12.5.1 什麽是索引文件 ···············.684
12.5.2 構建索引文件 ··················.685
12.5.3 利用索引文件提高查詢消息的性能 ································.687
12.6 提升消費消息RPC 請求的性能——利用零拷貝實現 ·························.687
12.6.1 什麽是零拷貝 ··················.688
12.6.2 使用Netty 的FileRegion 類實現文件的零拷貝 ····················.688
12.6.3 在Broker Server 的消息處理器中使用零拷貝 ·······················.691
12.7 提升Broker Server 寫消息的性能——利用內存映射、Page Cache 及堆外
內存實現 ··································.692
12.7.1 什麽是內存映射、Page Cache 和堆外內存 ··························.693
12.7.2 寫消息高性能的邏輯架構 ···.695
12.7.3 “利用MappedFile 文件實現內存映射”的原理 ··············.696
12.7.4 “利用內存映射實現Page Cache”的原理 ·····························.696
12.7.5 “管理堆外內存和Page Cache”的原理 ·····························.697
12.8 清理過期的ConsumeQueue 文件和CommitLog 文件 ························.702
12.8.1 清理過期的ConsumeQueue文件 ································.702
12.8.2 清理過期的CommitLog文件 ································.704
12.8.3 【實例】利用Spring CloudAlibaba 項目演示清理過期的ConsumeQueue 文件和CommitLog文件 ································.709
第5 篇 應用
第13 章 RocketMQ 在分佈式架構中的應用 ····································.712
13.1 RocketMQ 在Spring Boot 中的應用 ········································.712
13.1.1 為什麽RocketMQ 要支持Spring Boot ·······················.712
13.1.2 利用Spring Boot 作為基礎框架生產和消費消息 ·················.713
13.2 RocketMQ 在Spring Cloud Alibaba 中的應用 ·······································.719
13.2.1 為什麽Spring Cloud Alibaba 要支持RocketMQ ··················.719
13.2.2 利用Spring Cloud Alibaba 作為基礎框架生產和消費消息 ·····.719
13.2.3 Spring Cloud Alibaba 定製化RocketMQ 的原理 ···············.728
13.3 【實例】將命令式編程模式升級為函數式編程模式，完成消息的生產和消費 ········································.730
13.3.1 採用命令式編程模式生產和消費消息 ··························.730

13.3.2 將命令式編程模式升級為函數式編程模式 ·················.731
13.4 RocketMQ 在RocketMQ Connect 中的應用 ········································.732
13.4.1 什麽是Message Connect ·····.733
13.4.2 什麽是RocketMQ Connect ··.735
13.4.3 RocketMQ Connect 的核心架構 ································.736
13.4.4 “利用生產者和消費者來同步集群信息”的原理 ··············.739
13.4.5 “利用監聽器完成服務發現”的原理 ································.743
13.4.6 “利用客戶端負載均衡實現同步消息的高吞吐量”的原理 ·····.744
13.4.7 【實例】搭建RocketMQ Connect的本地和分佈式環境 ···········.747
13.5 RocketMQ 在Event Bridge 中的應用 ········································.750
13.5.1 什麽是Event Bridge ···········.750
13.5.2 Event Bridge 的核心架構 ·····.751
13.5.3 利用RocketMQ 實現EventBridge 的核心原理 ··············.756
13.6 RocketMQ 在多語言環境中的應用 ········································.757
13.6.1 RocketMQ 的多語言架構設計 ································.757
13.6.2 【實例】使用Go 語言客戶端生產和消費消息 ·················.758
13.7 RocketMQ 在大數據中的應用 ······.760
13.7.1 RocketMQ Streams 的核心架構 ································.761
13.7.2 【實例】利用RocketMQ Streams處理微服務中的流數據 ········.765
第6 篇 新特性
章 分析RocketMQ 5.x 的新特性 ··.768
14.1 利用“對等部署模式”實現BrokerServer 的高可用 ·························.768
14.1.1 認識“對等部署模式” ······.768
14.1.2 實現“對等部署模式”的原理 ································.771
14.1.3 【實例】利用微服務來驗證“對等部署模式”的高可用 ···.777
14.2 利用“Slave 代理Master 模式”實現Broker Server 的高可用 ················.778
14.2.1 認識“Slave 代理Master模式” ·····························.778
14.2.2 實現“Slave 代理Master模式”的原理 ····················.779
14.2.3 【實例】利用微服務來驗證“Slave 代理Master 模式”的
高可用 ·····························.788
14.3 利用“RocketMQ Controller 模式”實現Broker Server 的高可用 ·········.790
14.3.1 認識“RocketMQ Controller模式” ·····························.791
14.3.2 實現“RocketMQ Controller模式”的原理 ····················.793
14.3.3 利用“RocketMQ Controller模式”完成集群管理 ···········.797
14.3.4 實現“RocketMQ Controller 模式中多副本狀態機”的原理 ·····.806
14.3.5 【實例】驗證“RocketMQController 模式”的
高可用性 ··························.810
14.4 利用“RocketMQ Proxy 模式”統一多語言客戶端 ····························.814
14.4.1 認識“RocketMQ Proxy模式” ·····························.814
14.4.2 實現“RocketMQ Proxy 模式”的原理 ································.816
14.4.3 實現“代理多語言客戶端請求”的原理 ···························.821