RabbitMQ都寫了，RocketMQ怎么能落下？
整體架構最近看到了我在Github上寫的rabbitmq-examples陸續被人star了，就想著寫個rocketmq-examples。對rabbitmq感興趣的小伙伴可以看我之前的文章。下面把RocketMQ的各個特性簡單介紹一下，這樣在用的時候心里也更有把握。
全網最全RabbitMQ總結，別再說你不會RabbitMQ
RocketMQ是阿里自研的消息中間件，RocketMQ的整體架構如下
主要有4個角色
Producer：消息生產者。類似，發信者 Consumer：消息消費者。類似，收信者 BrokerServer：消息的存儲，投遞，查詢。類似，郵局 NameServer：注冊中心，支持Broker的動態注冊與發現。類似，郵局的管理結構
再介紹幾個基本概念
Topic(主題)：一類消息的集合，Topic和消息是一對多的關系。每個Broker可以存儲多個Topic的消息，每個Topic也可以分片存儲于不同的Broker
Tag(標簽)：在Topic類別下的二級子類別。如財務系統的所有消息的Topic為Finance_Topic，創建訂單消息的Tag為Create_Tag，關閉訂單消息的Tag為Close_Tag。這樣就能根據Tag消費不同的消息，當然你也可以為創建訂單和關閉訂單的消息各自創建一個Topic
Message Queue(消息隊列)：相當于Topic的分區，用于并行發送和消費消息。Message Queue在Broker上，一個Topic默認的Message Queue的數量為4
Producer Group(生產者組)：同一類Producer的集合。如果發送的是事務消息且原始生產者在發送之后崩潰，Broker會聯系統一生產者組內的其他生產者實例以提交或回溯消費
Consumer Group(消費者組)：同一類Consumer的集合。消費者組內的實例必須訂閱完全相同的Topic
Clustering(集群消費)：相同Consumer Group下的每個Consumer實例平均分攤消息
Broadcasting(廣播消費)：相同Consumer Group的每個Consumer實例都接收全量的消息
用圖演示一下Clustering和Broadcasting的區別
如果我有一條訂單程成交的消息，財務系統和物流系統都要同時訂閱消費這條消息，該怎么辦呢?定義2個Consumer Group即可
Consumer1和Consumer2屬于一個Consumer Group，Consumer3和Consumer4屬于一個Consumer Group，消息會全量發送到這2個Consuemr Group，至于這2個Consumer Group是集群消費還是廣播消費，自己定義即可
工作流程在官方文檔寫的很詳細，不再深入了
https://github.com/apache/rocketmq/tree/master/docs/cn
Message消息的各種處理方式涉及到的內容較多，所以我就不在文章中放代碼了，直接放GitHub了，目前還在不斷完善中
地址為：https://github.com/erlieStar/rocketmq-examples，
和之前的RabbitMQ一個風格，基本上所有知識點都涉及到了
地址為：https://github.com/erlieStar/rabbitmq-example
每個消息必須屬于一個Topic。RocketMQ中每個消息具有唯一的Message Id，且可以攜帶具有業務標識的Key，我們可以通過Topic，Message Id或Key來查詢消息
消息消費的方式
- Pull(拉取式消費)，Consumer主動從Broker拉取消息
- Push(推送式消費)，Broker收到數據后會主動推送給Consumer，實時性較高
消息的過濾方式
- 指定Tag
- SQL92語法過濾
消息的發送方式
- 同步，收到響應后才會發送下一條消息
- 異步，一直發，用異步的回調函數來獲取結果
- 單向(只管發，不管結果)
消息的種類
- 順序消息
- 延遲消息
- 批量消息
- 事務消息
順序消息
順序消息分為局部有序和全局有序
官方介紹為普通順序消息和嚴格順序消息
局部有序：同一個業務相關的消息是有序的，如針對同一個訂單的創建和付款消息是有序的，只需要在發送的時候指定message queue即可，如下所示，將同一個orderId對應的消息發送到同一個隊列
- SendResult sendResult = producer.send(message, new MessageQueueSelector() {
- /**
- * @param mqs topic對應的message queue
- * @param msg send方法傳入的message
- * @param arg send方法傳入的orderId
- */
- @Override
- public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
- // 根據業務對象選擇對應的隊列
- Integer orderId = (Integer) arg;
- int index = orderId % mqs.size();
- return mqs.get(index);
- }
- }, orderId);
消費者所使用的Listener必須是MessageListenerOrderly(對于一個隊列的消息采用一個線程去處理)，而平常的話我們使用的是MessageListenerConcurrently
全局有序：要想實現全局有序，則Topic只能有一個message queue。
延遲消息
RocketMQ并不支持任意時間的延遲，需要設置幾個固定的延時等級，從1s到2h分別對應著等級1到18
- // org.apache.rocketmq.store.config.MessageStoreConfig
- private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h"
批量消息
批量發送消息能顯著提高傳遞小消息的性能，限制是這批消息應該有相同的topic，相同的waitStoreMsgOK，而且不能是延時消息，一批消息的總大小不應超過1MB
事務消息
事務在實際的業務場景中還是經常遇到的，以轉賬為例子
張三給李四轉賬100元，可以分為如下2步
- 張三的賬戶減去100元
- 李四的賬戶加上100元
這2個操作要是同時成功，要是同時失敗，不然會造成數據不一致的情況，基于單個數據庫Connection時，我們只需要在方法上加上@Transactional注解就可以了。
如果基于多個Connection(如服務拆分，數據庫分庫分表)，加@Transactional此時就不管用了，就得用到分布式事務
分布式事務的解決方案很多，RocketMQ只是其中一種方案，RocketMQ可以保證最終一致性
RocketMQ實現分布式事務的流程如下
- producer向mq server發送一個半消息
- mq server將消息持久化成功后，向發送方確認消息已經發送成功，此時消息并不會被consumer消費
- producer開始執行本地事務邏輯
- producer根據本地事務執行結果向mq server發送二次確認，mq收到commit狀態，將消息標記為可投遞，consumer會消費該消息。mq收到rollback則刪除半消息，consumer將不會消費該消息，如果收到unknow狀態，mq會對消息發起回查
- 在斷網或者應用重啟等特殊情況下，步驟4提交的2次確認有可能沒有到達mq server，經過固定時間后mq會對該消息發起回查
- producer收到回查后，需要檢查本地事務的執行狀態
- producer根據本地事務的最終狀態，再次提交二次確認，mq仍按照步驟4對半消息進行操作
理解了原理，看代碼實現就很容易了，放一個官方的example
- public class TransactionListenerImpl implements TransactionListener {
- private AtomicInteger index = new AtomicInteger(0);
- private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();
- @Override
- public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
- int value = index.getAndIncrement();
- int status = value % 3;
- localTrans.put(msg.getTransactionId(), status);
- return LocalTransactionState.UNKNOW;
- }
- @Override
- public LocalTransactionState checkLocalTransaction(MessageExt msg) {
- Integer status = localTrans.get(msg.getTransactionId());
- if (status != null) {
- switch (status) {
- case 0:
- return LocalTransactionState.UNKNOW;
- case 1:
- return LocalTransactionState.COMMIT_MESSAGE;
- case 2:
- return LocalTransactionState.ROLLBACK_MESSAGE;
- default:
- return LocalTransactionState.COMMIT_MESSAGE;
- }
- }
- return LocalTransactionState.COMMIT_MESSAGE;
- }
- }
實現分布式事務需要實現TransactionListener接口，2個方法的作用如下
- executeLocalTransaction，執行本地事務
- checkLocalTransaction，回查本地事務狀態
針對這個例子，所有的消息都會回查，因為返回的都是UNKNOW，回查的時候status=1的數據會被消費，status=2的數據會被刪除，status=0的數據會一直回查，直到超過默認的回查次數。
發送方代碼如下
- public class TransactionProducer {
- public static final String RPODUCER_GROUP_NAME = "transactionProducerGroup";
- public static final String TOPIC_NAME = "transactionTopic";
- public static final String TAG_NAME = "transactionTag";
- public static void main(String[] args) throws Exception {
- TransactionListener transactionListener = new TransactionListenerImpl();
- TransactionMQProducer producer = new TransactionMQProducer(RPODUCER_GROUP_NAME);
- ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS,
- new ArrayBlockingQueue<>(100), new ThreadFactory() {
- @Override
- public Thread newThread(Runnable r) {
- Thread thread = new Thread();
- thread.setName("transaction-msg-check-thread");
- return thread;
- }
- });
- producer.setExecutorService(executorService);
- producer.setTransactionListener(transactionListener);
- producer.start();
- for (int i = 0; i < 100; i++) {
- Message message = new Message(TOPIC_NAME, TAG_NAME,
- ("hello rocketmq " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
- SendResult sendResult = producer.send(message);
- System.out.println(sendResult);
- }
- TimeUnit.HOURS.sleep(1);
- producer.shutdown();
- }
- }
看到這，可能有人會問了，我們先執行本地事務，執行成功后再發送消息，這樣可以嗎?
其實這樣做還是有可能會造成數據不一致的問題。假如本地事務執行成功，發送消息，由于網絡延遲，消息發送成功，但是回復超時了，拋出異常，本地事務回滾。但是消息其實投遞成功并被消費了，此時就會造成數據不一致的情況
那消息投遞到mq server，consumer消費失敗怎么辦?
如果是消費超時，重試即可。如果是由于代碼等原因真的消費失敗了，此時就得人工介入，重新手動發送消息，達到最終一致性。
消息重試
發送端重試
producer向broker發送消息后，沒有收到broker的ack時，rocketmq會自動重試。重試的次數可以設置，默認為2次
- DefaultMQProducer producer = new DefaultMQProducer(RPODUCER_GROUP_NAME);
- // 同步發送設置重試次數為5次
- producer.setRetryTimesWhenSendFailed(5);
- // 異步發送設置重試次數為5次
- producer.setRetryTimesWhenSendAsyncFailed(5);
消費端重試
順序消息的重試
對于順序消息，當Consumer消費消息失敗后，RocketMQ會不斷進行消息重試，此時后續消息會被阻塞。所以當使用順序消息的時候，監控一定要做好，避免后續消息被阻塞
無序消息的重試
當消費模式為集群模式時，Broker才會自動進行重試，對于廣播消息是不會進行重試的
當consumer消費消息后返回ConsumeConcurrentlyStatus.CONSUME_SUCCESS表明消費消息成功，不會進行重試
當consumer符合如下三種場景之一時，會對消息進行重試
- 返回ConsumeConcurrentlyStatus.RECONSUME_LATER
- 返回null
- 主動或被動拋出異常
RocketMQ默認每條消息會被重試16次，超過16次則不再重試，會將消息放到死信隊列,當然我們也可以自己設置重試次數
每次重試的時間間隔如下
| 第幾次重試 | 與上次間隔時間 | 第幾次重試 | 與上次間隔時間 |
|---|---|---|---|
| 1 | 10s | 10 | 7分鐘 |
| 2 | 30s | 11 | 8分鐘 |
| 3 | 1分鐘 | 12 | 9分鐘 |
| 4 | 2分鐘 | 13 | 10分鐘 |
| 5 | 3分鐘 | 14 | 20分鐘 |
| 6 | 4分鐘 | 15 | 30分鐘 |
| 7 | 5分鐘 | 16 | 1小時 |
| 8 | 6分鐘 | 17 | 2小時 |
重試隊列和死信隊列
當消息消費失敗，會被發送到重試隊列
當消息消費失敗，并達到最大重試次數，rocketmq并不會將消息丟棄，而是將消息發送到死信隊列
死信隊列有如下特點
- 里面存的是不能被正常消費的消息
- 有效期與正常消息相同，都是3天，3天后會被刪除
- 每個死信隊列對應一個Consumer Group ID，即死信隊列是消費者組級別的
- 如果一個Consumer Group沒有產生死信消息，則RocketMQ不會創建對應的死信隊列
- 死信隊列包含了一個Consumer Group下的所有死信消息，不管該消息屬于哪個Topic
- 重試隊列的命名為 %RETRY%消費組名稱 死信隊列的命名為 %DLQ%消費組名稱
RocketMQ高性能和高可用的方式
整體架構
rocketmq是通過broker主從機制來實現高可用的。相同broker名稱，不同brokerid的機器組成一個broker組，brokerId=0表明這個broker是master，brokerId>0表明這個broker是slave。
消息生產的高可用：創建topic時，把topic的多個message queue創建在多個broker組上。這樣當一個broker組的master不可用后，producer仍然可以給其他組的master發送消息。rocketmq目前還不支持主從切換，需要手動切換
消息消費的高可用：consumer并不能配置從master讀還是slave讀。當master不可用或者繁忙的時候consumer會被自動切換到從slave讀。這樣當master出現故障后，consumer仍然可以從slave讀，保證了消息消費的高可用
消息存儲結構
RocketMQ需要保證消息的高可靠性，所以要將數據通過磁盤進行持久化存儲。
將數據存到磁盤會不會很慢?其實磁盤有時候比你想象的快，有時候比你想象的慢。目前高性能磁盤的順序寫速度可以達到600M/s，而磁盤的隨機寫大概只有100k/s，和順序寫的性能相差6000倍，所以RocketMQ采用順序寫。
并且通過mmap(零拷貝的一種實現方式，零拷貝可以省去用戶態到內核態的數據拷貝，提高速度)具體原理并不是很懂，有興趣的小伙伴可以看看相關書籍
總而言之，RocketMQ通過順序寫和零拷貝技術實現了高性能的消息存儲
和消息相關的文件有如下幾種
- CommitLog：存儲消息的元數據
- ConsumerQueue：存儲消息在CommitLog的索引
- IndexFile：提供了一種通過key或者時間區間來查詢消息的方法
刷盤機制
- 同步刷盤：消息被寫入內存的PAGECACHE，返回寫成功狀態，當內存里的消息量積累到一定程度時，統一觸發寫磁盤操作，快速寫入 。吞吐量低，但不會造成消息丟失
- 異步刷盤：消息寫入內存的PAGECACHE后，立刻通知刷盤線程刷盤，然后等待刷盤完成，刷盤線程執行完成后喚醒等待的線程，給應用返回消息寫成功的狀態。吞吐量高，當磁盤損壞時，會丟失消息
主從復制
如果一個broker有master和slave時，就需要將master上的消息復制到slave上，復制的方式有兩種
- 同步復制：master和slave均寫成功，才返回客戶端成功。maste掛了以后可以保證數據不丟失，但是同步復制會增加數據寫入延遲，降低吞吐量
- 異步復制：master寫成功，返回客戶端成功。擁有較低的延遲和較高的吞吐量，但是當master出現故障后，有可能造成數據丟失
負載均衡
Producer負載均衡
producer在發送消息時，默認輪詢所有queue，消息就會被發送到不同的queue上。而queue可以分布在不同的broker上
Consumer負載均衡
默認的分配算法是AllocateMessageQueueAveragely，如下圖
還有另外一種平均的算法是AllocateMessageQueueAveragelyByCircle，也是平均分攤queue，只是以環狀輪流分queue的形式，如下圖：
如果consumer數量比message queue還多，則多會來的consumer會被閑置。所以不要讓consumer的數量多于message queue的數量
圖形化管理工具
在rocketmq-externals這個項目中提供了rocketmq的很多擴展工具
github地址如下：https://github.com/apache/rocketmq-externals
其中有一個子項目rocketmq-console提供了rocketmq的圖像化工具，提供了很多實用的功能，如前面說的通過Topic，Message Id或Key來查詢消息，重新發送消息等，還是很方便的
本文轉載自微信公眾號「Java識堂」，可以通過以下二維碼關注。轉載本文請聯系Java識堂公眾號。
