必須先理解的RocketMQ入門手冊，才能再次深入解讀
RocketMQ是一個純java、分布式、隊列模型的開源消息中間件，前身是MetaQ，是阿里研發的一個隊列模型的消息中間件，后開源給apache基金會成為了apache的頂級開源項目，具有高性能、高可靠、高實時、分布式特點，
同時，廣泛應用于多個領域，包括異步通信解耦、企業解決方案、金融支付、電信、電子商務、快遞物流、廣告營銷、社交、即時通信、移動應用、手游、視頻、物聯網、車聯網等。
具有以下特點：
- 能夠保證嚴格的消息順序
- 提供豐富的消息拉取模式
- 高效的訂閱者水平擴展能力
- 實時的消息訂閱機制
- 億級消息堆積能力
RocketMQ 架構原理分析
RocketMQ 架構
NameServer (名稱服務器)：
- 提供輕量級的服務發現和路由。NameServer接受來自Broker群集的注冊，并提供檢測信號機制以檢查Broker是否還存在
- 每個NameServer記錄完整的路由信息(Broker 相關 Topic 等元信息，并給 Producer 提供 Consumer 查找 Broker 信息)，提供相應的讀寫服務。
Broker(消息服務器)： 消息存儲中心，接收來自 Producer 的消息并存儲， Consumer 從這里取得消息
- 單個Broker節點與所有的NameServer節點保持長連接及心跳，并會定時將Topic信息注冊到NameServer，(其底層通信是基于Netty實現的)
- Broker負責消息存儲，以Topic為維度支持輕量級的隊列，單機可以支撐上萬隊列規模，支持消息推拉模型。
- 具有上億級消息堆積能力，同時可嚴格保證消息的有序性
Producer (生產者)：
- 負責產生消息，生產者向消息服務器發送由業務應用程序系統生成的消息
- 生產者支持分布式部署。分布式生產者通過多種負載平衡模式將消息發送到Broker集群。發送過程支持快速失敗并且延遲低
- 三種方式發送消息：同步、異步和單向
Consumer(消費者)：
- 負責消費消息，消費者從消息服務器拉取信息并將其輸入用戶應用程序
- 也支持“推和拉”模型中的分布式部署。
- 它還支持集群使用和消息廣播。它提供了實時消息訂閱機制，可以滿足大多數消費者的需求。
Broker Server
Broker Server負責消息的存儲和傳遞，消息查詢，HA高可用等，Broker Server幾個主要模塊組成：
Remoting Module(遠程模塊)：broker入口，處理來自客戶端的請求
Client Manager(客戶端管理)：管理client(生產者/消費者)并維護消費者的主題訂閱
Store Service(存儲服務)：提供簡單的API供給數據庫存儲或查詢消息
HA Service(高可用服務)：提供master broker和slave broker之間的數據同步功能
Index Service(索引服務)：將message建立索引來提供快速的查詢能力
RocketMQ 整體流程
1.啟動 NameServer，NameServer啟動后進行端口監聽，等待 Broker、Producer、Consumer 連上來，相當于一個路由控制中心
2.Broker 啟動，跟所有的 Namesrv 保持長連接，定時發送心跳包
- 心跳包中，包含當前 Broker 信息(IP+端口等)以及存儲所有 Topic 信息
- 注冊成功后，Namesrv 集群中就有 Topic 跟 Broker 的映射關系
3.收發消息前，先創建 Topic 。創建 Topic 時，需要指定該 Topic 要存儲在哪些 Broker上。也可以在發送消息時自動創建Topic
4.Producer 發送消息
- 啟動時，先跟 Namesrv 集群中的其中一臺建立長連接，并從Namesrv 中獲取當前發送的 Topic 存在哪些 Broker 上
- 然后跟對應的 Broker 建立長連接，直接向 Broker 發消息
5.Consumer 消費消息
- 跟其中一臺 Namesrv 建立長連接，獲取當前訂閱 Topic 存在哪些 Broker 上
- 然后直接跟 Broker 建立連接通道，開始消費消息*RocketMQ的消息領域模型
RocketMQ Message
Topic(主題)： 表示消息的第一級類型，是最細粒度的訂閱單位(生產者傳遞消息和消費者提取消息標識)
- 一條消息必須有一個Topic
- 一個Group可以訂閱多個Topic的消息
- Topic一般為領域范圍，比如交易消息
Tag(標簽)： 表示消息的第二級類型，可以是使用相同的Topic不同的Tag來表示同一業務模塊的不同任務的消息，比如交易消息又可以分為：交易創建消息，交易完成消息等
- 助于保持代碼整潔和一致
- 簡化RocketMQ提供的查詢系統
Message(消息體)： 消息是要傳遞的信息。Message中必須包含一個Topic，可選Tag和key-vaule鍵值對
Message Queue(消息隊列)： 所有消息隊列都是持久化
- 一個Topic下可以有多個Queue
- Queue的引入使得消息的存儲可以分布式集群化，具有了水平擴展能力
Group(組)： 分為Producer Group(生產者組)和Consumer Group(消費者組)，具有相同角色組成Group
- 原生產者在交易后崩潰，broker可以聯系同一生產者組的不同生產者實例以進行提交或回退交易。
- 消費者組的消費者實例必須具有完全相同的主題訂閱
RocketMQ 特性
Message Model(消息模式)：
- Clustering(集群式)：當使用集群消費模式時，MQ 認為任意一條消息只需要被集群內的任意一個消費者處理即可
- Broadcasting(廣播式)：當使用廣播消費模式時，MQ 會將每條消息推送給集群內所有注冊過的客戶端，保證消息至少被每臺機器消費一次
Message Order(消息順序)
- 使用DefaultMQPushConsumer時，可以決定按順序或同時使用消息
如果指定按順序使用，則消息使用的最大并發度是使用者組訂閱的消息隊列數
在此模式下不再保證消息順序
- Concurrently：同時使用消息時，消息使用的最大并發性僅受為每個使用方客戶端指定的線程池限制
- Orderly：有序地使用消息意味著消息的消費順序與生產者為每個消息隊列發送消息的順序相同。( 如果要處理必須強制執行全局順序的情況，請確保您使用的主題只有一個消息隊列)
Message Types(消息類型)
- 事務消息
- 順序消息
- 延遲消息
RocketMQ單機版安裝
1.下載編譯源碼
- # 下載$
- > wget wget http://mirror.bit.edu.cn/apache/rocketmq/4.6.0/rocketmq-all-4.6.0-source- >
- # 解壓$
- >unzip rocketmq-all-4.7.0-source-release.zip
- > cd rocketmq-all-4.7.0/
- # 編譯$
- > mvn -Prelease-all -DskipTests clean install -U
- > cd distribution/target/rocketmq-4.7.0/rocketmq-4.7.0
2.啟動 Name Server
- # 啟動 Name Server 服務
- > nohup sh bin/mqnamesrv &
- # 啟動完成后，查看日志$
- > tail -f ~/logs/rocketmqlogs/namesrv.log
- The Name Server boot success...
3.啟動 Broker
在 conf 目錄下，RocketMQ 提供了多種 Broker 的配置文件：
- # 啟動 Broker服務
- > nohup sh bin/mqbroker -n localhost:9876 &
- # 啟動完成后，查看日志$
- > tail -f ~/logs/rocketmqlogs/broker.log
- The broker[%s, 172.30.30.233:10911] boot success...
其中，參數：
- 通過 -c 參數，配置讀取的主 Broker 配置
- 通過 -n 參數，設置 RocketMQ Namesrv 地址
- broker.conf ：單主，異步刷盤。
- 2m/ ：雙主，異步刷盤。
- 2m-2s-async/ ：兩主兩從，異步復制，異步刷盤。
- 2m-2s-sync/ ：兩主兩從，同步復制，異步刷盤。
- dledger/ ：Dledger 集群，至少三節點
4.Send & Receive Messages(消息發送與接收)
在發送/接收消息之前，我們需要告知client(生產者/消費者)Name Servers的地址。RocketMQ提供了多種方法來實現：
為簡單起見，我們使用環境變量：NAMESRV_ADDR，如下所示：
- # 設置 Name Servers的地址$
- > export NAMESRV_ADDR=localhost:9876
- # 生產消息$
- > sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer
- SendResult [sendStatus=SEND_OK, msgId= ...
- # 消費消息$
- > sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer
- ConsumeMessageThread_%d Receive New Messages: [MessageExt...
- 在代碼中設置：producer.setNamesrvAddr("ip:port")
- java屬性配置：rocketmq.namesrv.addr
- 環境變量配置：NAMESRV_ADDR
- HTTP Endpoint
