可以發(fā)現(xiàn)，RocketMQ是把Topic分片存儲到各個(gè)Broker節(jié)點(diǎn)中，然后在把Broker節(jié)點(diǎn)中的Topic繼續(xù)分片為若干等分的ConsumeQueue，從而提高消息的吞吐量。ConsumeQueue是作為負(fù)載均衡資源分配的基本單元。

這樣把Topic的消息分區(qū)到了不同的Broker上，從而增加了消息隊(duì)列的數(shù)量，從而能夠支持更塊的并發(fā)消費(fèi)速度(只要有足夠的消費(fèi)者)。

4.3.2 Broker自動創(chuàng)建Topic會有什么問題？

假設(shè)設(shè)置為通過Broker自動創(chuàng)建Topic(autoCreateTopicEnable=true)，并且Producer端設(shè)置Topic消息隊(duì)列數(shù)量設(shè)置為4，也就是默認(rèn)值：

producer.setDefaultTopicQueueNums(4);

嘗試往一個(gè)新的 topic itzhai-test-queue-1連續(xù)發(fā)送10條消息，發(fā)送完畢之后，查看Topic狀態(tài)：

我們可以發(fā)現(xiàn)，在兩個(gè)broker上面都創(chuàng)建了itzhai-test-queue-a，并且每個(gè)broker上的消息隊(duì)列數(shù)量都為4。怎么回事，我配置的明明是期望創(chuàng)建4個(gè)隊(duì)列，為什么加起來會變成了8個(gè)？如下圖所示：

由于時(shí)間關(guān)系，本文我們不會帶大家從源碼方面去解讀為啥會出現(xiàn)這種情況，接下來我們通過一種更加直觀的方式來驗(yàn)證下這個(gè)問題：繼續(xù)做實(shí)驗(yàn)。

我們繼續(xù)嘗試往一個(gè)新的 topic itzhai-test-queue-10發(fā)送1條消息，注意，這一次不做并發(fā)發(fā)送了，只發(fā)送一條，發(fā)送完畢之后，查看Topic狀態(tài)：

可以發(fā)現(xiàn)，這次創(chuàng)建的消息隊(duì)列數(shù)量又是對的了，并且都是在broker-a上面創(chuàng)建的。接下來，無論怎么并發(fā)發(fā)送消息，消息隊(duì)列的數(shù)量都不會繼續(xù)增加了。

其實(shí)這也是并發(fā)請求Broker，觸發(fā)自動創(chuàng)建Topic的bug。

為了更加嚴(yán)格的管理Topic的創(chuàng)建和分片配置，一般在生產(chǎn)環(huán)境都是配置為手動創(chuàng)建Topic，通過提交運(yùn)維工單申請創(chuàng)建Topic以及Topic的數(shù)據(jù)分配。

接下來我們來看看RocketMQ的特性。更多其他技術(shù)的底層架構(gòu)內(nèi)幕分析，請?jiān)L問我的博客IT宅(itzhai.com)或者關(guān)注Java架構(gòu)雜談公眾號。

5. RocketMQ特性

5.1 生產(chǎn)端

5.1.1 消息發(fā)布

RocketMQ中定義了如下三種消息通信的方式：

public enum CommunicationMode {    SYNC,    ASYNC,    ONEWAY,}

• SYNC：同步發(fā)送，生產(chǎn)端會阻塞等待發(fā)送結(jié)果；
  • 應(yīng)用場景：這種方式應(yīng)用場景非常廣泛，如重要業(yè)務(wù)事件通知。
• ASYNC：異步發(fā)送，生產(chǎn)端調(diào)用發(fā)送API之后，立刻返回，在拿到Broker的響應(yīng)結(jié)果后，觸發(fā)對應(yīng)的SendCallback回調(diào)；
  • 應(yīng)用場景：一般用于鏈路耗時(shí)較長，對 RT 較為敏感的業(yè)務(wù)場景；
• ONEWAY：單向發(fā)送，發(fā)送方只負(fù)責(zé)發(fā)送消息，不等待服務(wù)器回應(yīng)且沒有回調(diào)函數(shù)觸發(fā)，即只發(fā)送請求不等待應(yīng)答。 此方式發(fā)送消息的過程耗時(shí)非常短，一般在微秒級別；
  • 應(yīng)用場景：適用于耗時(shí)非常短，對可靠性要求不高的場景，如日志收集。

SYNC和ASYNC關(guān)注發(fā)送結(jié)果，ONEWAY不關(guān)注發(fā)送結(jié)果。發(fā)送結(jié)果如下：

public enum SendStatus {    SEND_OK,    FLUSH_DISK_TIMEOUT,    FLUSH_SLAVE_TIMEOUT,    SLAVE_NOT_AVAILABLE,}

• SEND_OK：消息發(fā)送成功。SEND_OK并不意味著投遞是可靠的，要確保消息不丟失，需要開啟SYNC_MASTER同步或者SYNC_FLUSH同步寫；
• FLUSH_DISK_TIMEOUT：消息發(fā)送成功，但是刷盤超時(shí)。如果Broker的flushDiskType=SYNC_FLUSH，并且5秒內(nèi)沒有完成消息的刷盤，則會返回這個(gè)狀態(tài)；
• FLUSH_SLAVE_TIMEOUT：消息發(fā)送成功，但是服務(wù)器同步到Slave時(shí)超時(shí)。如果Broker的brokerRole=SYNC_MASTER，并且5秒內(nèi)沒有完成同步，則會返回這個(gè)狀態(tài)；
• SLAVE_NOT_AVAILABLE：消息發(fā)送成功，但是無可用的Slave節(jié)點(diǎn)。如果Broker的brokerRole=SYNC_MASTER，但是沒有發(fā)現(xiàn)SLAVE節(jié)點(diǎn)或者SLAVE節(jié)點(diǎn)掛掉了，那么會返回這個(gè)狀態(tài)。

源碼內(nèi)容更精彩，歡迎大家進(jìn)一步閱讀源碼詳細(xì)了解消息發(fā)送的內(nèi)幕：

• 同步發(fā)送：org.apache.rocketmq.client.producer.DefaultMQProducer#send(org.apache.rocketmq.common.message.Message)
• 異步發(fā)送：org.apache.rocketmq.client.producer.DefaultMQProducer#send(org.apache.rocketmq.common.message.Message, org.apache.rocketmq.client.producer.SendCallback)
• 單向發(fā)送：org.apache.rocketmq.client.producer.DefaultMQProducer#sendOneway(org.apache.rocketmq.common.message.Message)

5.1.2 順序消費(fèi)

消息的有序性指的是一類消息消費(fèi)的時(shí)候，可以按照發(fā)送順序來消費(fèi)，比如：在Java架構(gòu)雜談茶餐廳吃飯產(chǎn)生的消息：進(jìn)入餐廳、點(diǎn)餐、下單、上菜、付款，消息要按照這個(gè)順序消費(fèi)才有意義，但是多個(gè)顧客產(chǎn)生的消息是可以并行消費(fèi)的。順序消費(fèi)又分為全局順序消費(fèi)和分區(qū)順序消費(fèi)：

• 全局順序：同一個(gè)Topic下的消息，所有消息按照嚴(yán)格的FIFO順序進(jìn)行發(fā)布和消費(fèi)。適用于：性能要求不高，所有消息嚴(yán)格按照FIFO進(jìn)行發(fā)布和消費(fèi)的場景；
• 分區(qū)順序：同一個(gè)Topic下，根據(jù)消息的特定業(yè)務(wù)ID進(jìn)行sharding key分區(qū)，同一個(gè)分區(qū)內(nèi)的消息按照嚴(yán)格的FIFO順序進(jìn)行發(fā)布和消費(fèi)。適用于：性能要求高，在同一個(gè)分區(qū)中嚴(yán)格按照FIFO進(jìn)行發(fā)布和消費(fèi)的場景。

一般情況下，生產(chǎn)者是會以輪訓(xùn)的方式把消息發(fā)送到Topic的消息隊(duì)列中的：

在同一個(gè)Queue里面，消息的順序性是可以得到保證的，但是如果一個(gè)Topic有多個(gè)Queue，以輪訓(xùn)的方式投遞消息，那么就會導(dǎo)致消息亂序了。

為了保證消息的順序性，需要把保持順序性的消息投遞到同一個(gè)Queue中。

5.1.2.1 如何保證消息投遞的順序性

RocketMQ提供了MessageQueueSelector接口，可以用來實(shí)現(xiàn)自定義的選擇投遞的消息隊(duì)列的算法：

for (int i = 0; i < orderList.size(); i++) {    String content = "Hello itzhai.com. Java架構(gòu)雜談," + new Date();    Message msg = new Message("topic-itzhai-com", tags[i % tags.length], "KEY" + i,            content.getBytes(RemotingHelper.DEFAULT_CHARSET));    SendResult sendResult = producer.send(msg, new MessageQueueSelector() {        @Override        public MessageQueue select(List mqs, Message msg, Object arg) {            Long orderId = (Long) arg;            // 訂單號與消息隊(duì)列個(gè)數(shù)取模，保證讓同一個(gè)訂單號的消息落入同一個(gè)消息隊(duì)列            long index = orderId % mqs.size();            return mqs.get((int) index);        }    }, orderList.get(i).getOrderId());    System.out.printf("content: %s, sendResult: %s%n", content, sendResult);}

如上圖，我們實(shí)現(xiàn)了MessageQueueSelector接口，并在實(shí)現(xiàn)的select方法里面，指定了選擇消息隊(duì)列的算法：訂單號與消息隊(duì)列個(gè)數(shù)取模，保證讓同一個(gè)訂單號的消息落入同一個(gè)消息隊(duì)列：

有個(gè)異常場景需要考慮：假設(shè)某一個(gè)Master節(jié)點(diǎn)掛掉了，導(dǎo)致Topic的消息隊(duì)列數(shù)量發(fā)生了變化，那么繼續(xù)使用以上的選擇算法，就會導(dǎo)致在這個(gè)過程中同一個(gè)訂單的消息會分散到不同的消息隊(duì)列里面，最終導(dǎo)致消息不能順序消費(fèi)。

為了避免這種情況，只能選擇犧牲failover特性了。

現(xiàn)在投遞到消息隊(duì)列中的消息保證了順序，那如何保證消費(fèi)也是順序的呢？

5.1.2.2 如何保證消息消費(fèi)的順序性？

RocketMQ中提供了MessageListenerOrderly，該對象用于有順序收異步傳遞的消息，一個(gè)隊(duì)列對應(yīng)一個(gè)消費(fèi)線程，使用方法如下：

consumer.registerMessageListener(new MessageListenerOrderly() {    // 消費(fèi)次數(shù)，用于輔助模擬各種消費(fèi)結(jié)果    AtomicLong consumeTimes = new AtomicLong(0);    @Override    public ConsumeOrderlyStatus consumeMessage(List msgs, ConsumeOrderlyContext context) {        context.setAutoCommit(true);        System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);        this.consumeTimes.incrementAndGet();        if ((this.consumeTimes.get() % 2) == 0) {            return ConsumeOrderlyStatus.SUCCESS;        } else if ((this.consumeTimes.get() % 3) == 0) {            return ConsumeOrderlyStatus.ROLLBACK;        } else if ((this.consumeTimes.get() % 4) == 0) {            return ConsumeOrderlyStatus.COMMIT;        } else if ((this.consumeTimes.get() % 5) == 0) {            context.setSuspendCurrentQueueTimeMillis(3000);            return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;        }        return ConsumeOrderlyStatus.SUCCESS;    }});

如果您使用的是MessageListenerConcurrently，表示并發(fā)消費(fèi)，為了保證消息消費(fèi)的順序性，需要設(shè)置為單線程模式。

使用MessageListenerOrderly的問題：如果遇到某條消息消費(fèi)失敗，并且無法跳過，那么消息隊(duì)列的消費(fèi)進(jìn)度就會停滯。

5.1.3 延遲隊(duì)列(定時(shí)消息)

定時(shí)消費(fèi)是指消息發(fā)送到Broker之后不會立即被消費(fèi)，而是等待特定的時(shí)間之后才投遞到Topic中。定時(shí)消息會暫存在名為SCHEDULE_TOPIC_XXXX的topic中，并根據(jù)delayTimeLevel存入特定的queue，queueId=delayTimeLevel-1，一個(gè)queue只存相同延遲的消息，保證具有相同延遲的消息能夠順序消費(fèi)。比如，我們設(shè)置1秒后把消息投遞到topic-itzhai-comtopic，則存儲的文件目錄如下所示：

Broker會調(diào)度地消費(fèi)SCHEDULE_TOPIC_XXXX，將消息寫入真實(shí)的topic。

定時(shí)消息的副作用：定時(shí)消息會在第一次寫入Topic和調(diào)度寫入實(shí)際的topic都會進(jìn)行計(jì)數(shù)，因此發(fā)送數(shù)量，tps都會變高。

使用延遲隊(duì)列的場景：提交了訂單之后，如果等待超過約定的時(shí)間還未支付，則把訂單設(shè)置為超時(shí)狀態(tài)。

RocketMQ提供了以下幾個(gè)固定的延遲級別：

public class MessageStoreConfig {    ...    // 10個(gè)level，level：1~18    private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";    ...}

level = 0 表示不使用延遲消息。

另外，消息消費(fèi)失敗也會進(jìn)入延遲隊(duì)列，消息發(fā)送時(shí)間與設(shè)置的延遲級別和重試次數(shù)有關(guān)。

以下是發(fā)送延遲消息的代碼：

public class ScheduledMessageProducer {    public static void main(String[] args) throws Exception {        DefaultMQProducer producer = new DefaultMQProducer("TestProducerGroup");        producer.start();        int totalMessagesToSend = 100;        for (int i = 0; i < totalMessagesToSend; i++) {            Message message = new Message("TestTopic", ("Hello scheduled message " + i).getBytes());            // 指定該消息在10秒后被消費(fèi)者消費(fèi)            message.setDelayTimeLevel(3);            producer.send(message);        }        producer.shutdown();    }}

5.1.4 數(shù)據(jù)完整性與事務(wù)消息

通過消息對系統(tǒng)進(jìn)行解耦之后，勢必會遇到分布式系統(tǒng)數(shù)據(jù)完整性的問題。

5.1.4.1 實(shí)現(xiàn)分布式事務(wù)的手段有哪些？

我們可以通過以下手段解決分布式系統(tǒng)數(shù)據(jù)最終一致性問題：

• 數(shù)據(jù)庫層面的2PC(Two-phase commit protocol)，二階段提交，同步阻塞，效率低下，存在協(xié)調(diào)者單點(diǎn)故障問題，極端情況下存在數(shù)據(jù)不一致的風(fēng)險(xiǎn)。對應(yīng)技術(shù)上的XA、JTA/JTS。這是分布式環(huán)境下事務(wù)處理的典型模式；
• 數(shù)據(jù)庫層面的3PC，三階段提交，引入了參與者超時(shí)機(jī)制，增加了預(yù)提交階段，使得故障恢復(fù)之后協(xié)調(diào)者的決策復(fù)雜度降低，但整體的交互過程變得更長了，性能有所下降，仍舊會存在數(shù)據(jù)不一致的問題；
• 業(yè)務(wù)層面的TCC ，Try - Confirm - Cancel。對業(yè)務(wù)的侵入較大，和業(yè)務(wù)緊耦合，對于每一個(gè)操作都需要定義三個(gè)動作分別對應(yīng)：Try - Confirm - Cancel，將資源層的兩階段提交協(xié)議轉(zhuǎn)換到業(yè)務(wù)層，成為業(yè)務(wù)模型中的一部分；
• 本地消息表；
• 事務(wù)消息；

RocketMQ事務(wù)消息（Transactional Message）則是通過事務(wù)消息來實(shí)現(xiàn)分布式事務(wù)的最終一致性。下面看看RocketMQ是如何實(shí)現(xiàn)事務(wù)消息的。

5.1.4.2 RocketMQ如何實(shí)現(xiàn)事務(wù)消息？

如下圖：

事務(wù)消息有兩個(gè)流程：

1. 事務(wù)消息發(fā)送及提交：
   1. 發(fā)送half消息；
   2. 服務(wù)端響應(yīng)half消息寫入結(jié)果；
   3. 根據(jù)half消息的發(fā)送結(jié)果執(zhí)行本地事務(wù)。如果發(fā)送失敗，此時(shí)half消息對業(yè)務(wù)不可見，本地事務(wù)不執(zhí)行；
   4. 根據(jù)本地事務(wù)狀態(tài)執(zhí)行Commit或者Rollback。Commit操作會觸發(fā)生成ConsumeQueue索引，此時(shí)消息對消費(fèi)者可見；
2. 補(bǔ)償流程：
   5. 對于沒有Commit/Rollback的事務(wù)消息，會處于pending狀態(tài)，這對這些消息，MQ Server發(fā)起一次回查；
   6. Producer收到回查消息，檢查回查消息對應(yīng)的本地事務(wù)的轉(zhuǎn)塔體；
   7. 根據(jù)本地事務(wù)狀態(tài)，重新執(zhí)行Commit或者Rollback。

補(bǔ)償階段主要用于解決消息的Commit或者Rollback發(fā)生超時(shí)或者失敗的情況。

half消息：并不是發(fā)送了一半的消息，而是指消息已經(jīng)發(fā)送到了MQ Server，但是該消息未收到生產(chǎn)者的二次確認(rèn)，此時(shí)該消息暫時(shí)不能投遞到具體的ConsumeQueue中，這種狀態(tài)的消息稱為half消息。

5.1.4.3 RocketMQ事務(wù)消息是如何存儲的？

發(fā)送到MQ Server的half消息對消費(fèi)者是不可見的，為此，RocketMQ會先把half消息的Topic和Queue信息存儲到消息的屬性中，然后把該half消息投遞到一個(gè)專門的處理事務(wù)消息的隊(duì)列中：RMQ_SYS_TRANS_HALF_TOPIC，由于消費(fèi)者沒有訂閱該Topic，所以無法消息half類型的消息。

生產(chǎn)者執(zhí)行Commit half消息的時(shí)候，會存儲一條專門的Op消息，用于標(biāo)識事務(wù)消息已確定的狀態(tài)，如果一條事務(wù)消息還沒有對應(yīng)的Op消息，說明這個(gè)事務(wù)的狀態(tài)還無法確定。RocketMQ會開啟一個(gè)定時(shí)任務(wù)，對于pending狀態(tài)的消息，會先向生產(chǎn)者發(fā)送回查事務(wù)狀態(tài)請求，根據(jù)事務(wù)狀態(tài)來決定是否提交或者回滾消息。

當(dāng)消息被標(biāo)記為Commit狀態(tài)之后，會把half消息的Topic和Queue相關(guān)屬性還原為原來的值，最終構(gòu)建實(shí)際的消費(fèi)索引(ConsumeQueue)。

RocketMQ并不會無休止的嘗試消息事務(wù)狀態(tài)回查，默認(rèn)查找15次，超過了15次還是無法獲取事務(wù)狀態(tài)，RocketMQ默認(rèn)回滾該消息。并打印錯(cuò)誤日志，可以通過重寫AbstractTransactionalMessageCheckListener類修改這個(gè)行為。

可以通過Broker的配置參數(shù)：transactionCheckMax來修改此值。

5.1.5 消息重投

如果消息發(fā)布方式是同步發(fā)送會重投，如果是異步發(fā)送會重試。

消息重投可以盡可能保證消息投遞成功，但是可能會造成消息重復(fù)。

什么情況會造成重復(fù)消費(fèi)消息？

• 出現(xiàn)消息量大，網(wǎng)絡(luò)抖動的時(shí)候；
• 生產(chǎn)者主動重發(fā)；
• 消費(fèi)負(fù)載發(fā)生變化。

可以使用的消息重試策略：

• retryTimesWhenSendFailed：設(shè)置同步發(fā)送失敗的重投次數(shù)，默認(rèn)為2。所以生產(chǎn)者最多會嘗試發(fā)送retryTimesWhenSendFailed+1次。
  • 為了最大程度保證消息不丟失，重投的時(shí)候會嘗試向其他broker發(fā)送消息；
  • 超過重投次數(shù)，拋出異常，讓客戶端自行處理；
  • 觸發(fā)重投的異常：RemotingException、MQClientException和部分MQBrokerException；
• retryTimesWhenSendAsyncFailed：設(shè)置異步發(fā)送失敗重試次數(shù)，異步重試不會選擇其他Broker，不保證消息不丟失；
• retryAnotherBrokerWhenNotStoreOK：消息刷盤(主或備)超時(shí)或slave不可用(返回狀態(tài)非SEND_OK)，是否嘗試發(fā)送到其他broker，默認(rèn)false。重要的消息可以開啟此選項(xiàng)。

oneway發(fā)布方式不支持重投。

5.1.6 批量消息

為了提高系統(tǒng)的吞吐量，提高發(fā)送效率，可以使用批量發(fā)送消息。

批量發(fā)送消息的限制：

• 同一批批量消息的topic，waitStoreMsgOK屬性必須保持一致；
• 批量消息不支持延遲隊(duì)列；
• 批量消息一次課發(fā)送的上限是4MB。

發(fā)送批量消息的例子：

String topic = "itzhai-test-topic";List messages = new ArrayList<>();messages.add(new Message(topic, "TagA", "OrderID001", "Hello world itzhai.com 0".getBytes()));messages.add(new Message(topic, "TagA", "OrderID002", "Hello world itzhai.com 1".getBytes()));messages.add(new Message(topic, "TagA", "OrderID003", "Hello world itzhai.com 2".getBytes()));producer.send(messages);

如果發(fā)送的消息比較多，會增加復(fù)雜性，為此，可以對大消息進(jìn)行拆分。以下是拆分的例子：

public class ListSplitter implements Iterator> {     // 限制最大大小    private final int SIZE_LIMIT = 1024 * 1024 * 4;    private final List messages;    private int currIndex;    public ListSplitter(List messages) {         this.messages = messages;    }    @Override public boolean hasNext() {        return currIndex < messages.size();     }    @Override public List next() {         int startIndex = getStartIndex();        int nextIndex = startIndex;        int totalSize = 0;        for (; nextIndex < messages.size(); nextIndex++) {            Message message = messages.get(nextIndex);             int tmpSize = calcMessageSize(message);            if (tmpSize + totalSize > SIZE_LIMIT) {                break;             } else {                totalSize += tmpSize;             }        }        List subList = messages.subList(startIndex, nextIndex);         currIndex = nextIndex;        return subList;    }    private int getStartIndex() {        Message currMessage = messages.get(currIndex);         int tmpSize = calcMessageSize(currMessage);         while(tmpSize > SIZE_LIMIT) {            currIndex += 1;            Message message = messages.get(curIndex);             tmpSize = calcMessageSize(message);        }        return currIndex;     }    private int calcMessageSize(Message message) {        int tmpSize = message.getTopic().length() + message.getBody().length();         Map properties = message.getProperties();        for (Map.Entry entry : properties.entrySet()) {            tmpSize += entry.getKey().length() + entry.getValue().length();         }        tmpSize = tmpSize + 20; // Increase the log overhead by 20 bytes        return tmpSize;     }}// then you could split the large list into small ones:ListSplitter splitter = new ListSplitter(messages);while (splitter.hasNext()) {   try {       List  listItem = splitter.next();       producer.send(listItem);   } catch (Exception e) {       e.printStackTrace();       // handle the error   }}

5.1.7 消息過濾

RocketMQ的消費(fèi)者可以根據(jù)Tag進(jìn)行消息過濾來獲取自己感興趣的消息，也支持自定義屬性過濾。

Tags是Topic下的次級消息類型/二級類型（注：Tags也支持TagA || TagB這樣的表達(dá)式），可以在同一個(gè)Topic下基于Tags進(jìn)行消息過濾。

消息過濾是在Broker端實(shí)現(xiàn)的，減少了對Consumer無用消息的網(wǎng)絡(luò)傳輸，缺點(diǎn)是增加了Broker負(fù)擔(dān)，實(shí)現(xiàn)相對復(fù)雜。

5.2 消費(fèi)端

5.2.1 消費(fèi)模型

消費(fèi)端有兩周消費(fèi)模型：集群消費(fèi)和廣播消費(fèi)。

集群消費(fèi)

集群消費(fèi)模式下,相同Consumer Group的每個(gè)Consumer實(shí)例平均分?jǐn)傁ⅰ?/p>

廣播消費(fèi)

廣播消費(fèi)模式下，相同Consumer Group的每個(gè)Consumer實(shí)例都接收全量的消息。

5.2.2 消息重試

RocketMQ會為每個(gè)消費(fèi)組都設(shè)置一個(gè)Topic名稱為%RETRY%consumerGroupName的重試隊(duì)列（這里需要注意的是，這個(gè)Topic的重試隊(duì)列是針對消費(fèi)組，而不是針對每個(gè)Topic設(shè)置的），用于暫時(shí)保存因?yàn)楦鞣N異常而導(dǎo)致Consumer端無法消費(fèi)的消息。

考慮到異常恢復(fù)起來需要一些時(shí)間，會為重試隊(duì)列設(shè)置多個(gè)重試級別，每個(gè)重試級別都有與之對應(yīng)的重新投遞延時(shí)，重試次數(shù)越多投遞延時(shí)就越大。

RocketMQ對于重試消息的處理是先保存至Topic名稱為SCHEDULE_TOPIC_XXXX的延遲隊(duì)列中，后臺定時(shí)任務(wù)按照對應(yīng)的時(shí)間進(jìn)行Delay后重新保存至%RETRY%consumerGroupName的重試隊(duì)列中。

比如，我們設(shè)置1秒后把消息投遞到topic-itzhai-comtopic，則存儲的文件目錄如下所示：

5.2.3 死信隊(duì)列

當(dāng)一條消息初次消費(fèi)失敗，消息隊(duì)列會自動進(jìn)行消息重試；達(dá)到最大重試次數(shù)后，若消費(fèi)依然失敗，則表明消費(fèi)者在正常情況下無法正確地消費(fèi)該消息，此時(shí)，消息隊(duì)列不會立刻將消息丟棄，而是將其發(fā)送到該消費(fèi)者對應(yīng)的特殊隊(duì)列中。

RocketMQ將這種正常情況下無法被消費(fèi)的消息稱為死信消息（Dead-Letter Message），將存儲死信消息的特殊隊(duì)列稱為死信隊(duì)列（Dead-Letter Queue）。

在RocketMQ中，可以通過使用console控制臺對死信隊(duì)列中的消息進(jìn)行重發(fā)來使得消費(fèi)者實(shí)例再次進(jìn)行消費(fèi)。

由于RocketMQ是使用Java寫的，所以它的代碼特別適合拿來閱讀消遣，我們繼續(xù)來看看RocketMQ的源碼結(jié)構(gòu)...

不不，還是算了，一下子又到周末晚上了，時(shí)間差不多了，今天就寫到這里了。有空再聊。

我精心整理了一份Redis寶典給大家，涵蓋了Redis的方方面面，面試官懂的里面有，面試官不懂的里面也有，有了它，不怕面試官連環(huán)問，就怕面試官一上來就問你Redis的Redo Log是干啥的？畢竟這種問題我也不會。

在Java架構(gòu)雜談公眾號發(fā)送Redis關(guān)鍵字獲取pdf文件：

本文作者： arthinking

博客鏈接： https://www.itzhai.com/articles/deep-understanding-of-rocketmq.html

高并發(fā)異步解耦利器：RocketMQ究竟強(qiáng)在哪里？

版權(quán)聲明： 版權(quán)歸作者所有，未經(jīng)許可不得轉(zhuǎn)載，侵權(quán)必究！聯(lián)系作者請加公眾號。

References

apache/rocketmq. Retrieved from https://github.com/apache/rocketmq