image.png

1) 同步刷盤

在返回寫成功狀態(tài)時(shí)汁汗，消息已經(jīng)被寫入磁盤。具體流程是栗涂，消息寫入內(nèi)存的PAGECACHE后知牌，立刻通知刷盤線程刷盤， 然后等待刷盤完成斤程，刷盤線程執(zhí)行完成后喚醒等待的線程角寸，返回消息寫 成功的狀態(tài)菩混。

2）異步刷盤

在返回寫成功狀態(tài)時(shí)，消息可能只是被寫入了內(nèi)存的PAGECACHE扁藕，寫操作的返回快沮峡，吞吐量大；當(dāng)內(nèi)存里的消息量積累到一定程度時(shí)亿柑，統(tǒng)一觸發(fā)寫磁盤動作邢疙，快速寫入。

3）配置

同步刷盤還是異步刷盤望薄，都是通過Broker配置文件里的flushDiskType 參數(shù)設(shè)置的疟游，這個(gè)參數(shù)被配置成SYNC_FLUSH、ASYNC_FLUSH中的一個(gè)痕支。

2 高可用機(jī)制

image.png

通過建立RocketMQ分布式集群達(dá)到高可用颁虐。

Broker中Master和Slave的區(qū)別：
在Broker的配置文件中，參數(shù) brokerId的值為 0 表明這個(gè)Broker是 Master 卧须，大于0表明這個(gè)Broker是 Slave另绩，同時(shí)brokerRole參數(shù)也會說明這個(gè)Broker是Master還是Slave。
master支持讀和寫故慈，slave僅支持讀板熊。

2.1 消息消費(fèi)高可用

在Consumer的配置文件中，并不需要設(shè)置是從Master讀還是從Slave 讀察绷，當(dāng)Master不可用或者繁忙的時(shí)候干签，Consumer會被自動切換到從Slave 讀。有了自動切換Consumer這種機(jī)制拆撼，當(dāng)一個(gè)Master角色的機(jī)器出現(xiàn)故障后容劳，Consumer仍然可以從Slave讀取消息，不影響Consumer程序闸度，從而達(dá)到了消費(fèi)端的高可用性竭贩。

2.2 消息發(fā)送高可用

在創(chuàng)建Topic的時(shí)候，把Topic的多個(gè)Message Queue創(chuàng)建在多個(gè)Broker組上（相同Broker名稱莺禁，不同 brokerId的機(jī)器組成一個(gè)Broker組）留量，這樣當(dāng)一個(gè)Broker組的Master不可用后，其他組的Master仍然可用哟冬，Producer仍然可以發(fā)送消息楼熄。 RocketMQ目前還不支持把Slave自動轉(zhuǎn)成Master，如果機(jī)器資源不足浩峡， 需要把Slave轉(zhuǎn)成Master可岂，則要手動停止Slave角色的Broker，更改配置文件翰灾，用新的配置文件啟動Broker缕粹。

2.3 消息主從復(fù)制

如果一個(gè)Broker組有Master和Slave稚茅，消息需要從Master復(fù)制到Slave 上，有同步和異步兩種復(fù)制方式平斩。

1）同步復(fù)制

同步復(fù)制方式是等Master和Slave均寫成功后才反饋給客戶端寫成功狀態(tài)亚享；
在同步復(fù)制方式下，如果Master出故障绘面， Slave上有全部的備份數(shù)據(jù)虹蒋，容易恢復(fù)，但是同步復(fù)制會增大數(shù)據(jù)寫入 延遲飒货，降低系統(tǒng)吞吐量。

2）異步復(fù)制

異步復(fù)制方式是只要Master寫成功即可反饋給客戶端寫成功狀態(tài)峭竣。
在異步復(fù)制方式下塘辅，系統(tǒng)擁有較低的延遲和較高的吞吐量，但是如果Master出了故障皆撩，有些數(shù)據(jù)因?yàn)闆]有被寫 入Slave扣墩，有可能會丟失；

3）配置

同步復(fù)制和異步復(fù)制是通過Broker配置文件里的brokerRole參數(shù)進(jìn)行設(shè)置的扛吞，這個(gè)參數(shù)可以被設(shè)置成ASYNC_MASTER呻惕、 SYNC_MASTER、SLAVE(slave機(jī)器設(shè)置成這個(gè)參數(shù))三個(gè)值中的一個(gè)滥比。

4）總結(jié)

image.png

實(shí)際應(yīng)用中要結(jié)合業(yè)務(wù)場景亚脆，合理設(shè)置刷盤方式和主從復(fù)制方式， 尤其是SYNC_FLUSH方式盲泛，由于頻繁地觸發(fā)磁盤寫動作濒持，會明顯降低性能。通常情況下寺滚，應(yīng)該把Master和Save配置成 ASYNC_FLUSH 的刷盤方式柑营，主從之間配置成 SYNC_MASTER 的復(fù)制方式，這樣即使有一臺機(jī)器出故障村视，仍然能保證數(shù)據(jù)不丟官套。

3 負(fù)載均衡

3.1 Producer負(fù)載均衡

Producer端，每個(gè)實(shí)例在發(fā)消息的時(shí)候蚁孔，默認(rèn)會 輪詢 所有的message queue發(fā)送奶赔，以達(dá)到讓消息平均落在不同的queue上。而由于queue可以散落在不同的broker勒虾，所以消息就發(fā)送到不同的broker下纺阔，如下圖：

默認(rèn)采用輪詢方式

圖中箭頭線條上的標(biāo)號代表順序，發(fā)布方會把第一條消息發(fā)送至 Queue 0修然，然后第二條消息發(fā)送至 Queue 1笛钝，以此類推质况。

3.2 Consumer負(fù)載均衡

1） 集群模式

在集群消費(fèi)模式下，每條消息只需要投遞到訂閱這個(gè)topic的Consumer Group下的一個(gè)實(shí)例即可玻靡。RocketMQ采用 主動拉取 的方式拉取并消費(fèi)消息结榄，在拉取的時(shí)候需要明確指定拉取哪一條message queue。

每當(dāng)實(shí)例的數(shù)量有變更囤捻，都會觸發(fā)一次所有實(shí)例的負(fù)載均衡臼朗，這時(shí)候會按照queue的數(shù)量和實(shí)例的數(shù)量平均分配queue給每個(gè)實(shí)例。

默認(rèn)的分配算法是AllocateMessageQueueAveragely蝎土，如下圖：

每個(gè)comsumer實(shí)例平均分配每個(gè)consume queue

還有另外一種平均的算法是AllocateMessageQueueAveragelyByCircle视哑，也是平均分?jǐn)偯恳粭lqueue，只是以環(huán)狀輪流分queue的形式誊涯，如下圖：

每個(gè)comsumer實(shí)例平均分配每個(gè)consume queue

ps: 集群模式下挡毅，queue都是只允許分配只 一個(gè)實(shí)例 ，這是由于如果多個(gè)實(shí)例同時(shí)消費(fèi)一個(gè)queue的消息暴构，由于拉取哪些消息是consumer主動控制的跪呈，那樣會導(dǎo)致同一個(gè)消息在不同的實(shí)例下被消費(fèi)多次，所以算法上都是一個(gè)queue只分給一個(gè)consumer實(shí)例取逾，一個(gè)consumer實(shí)例可以允許同時(shí)分到不同的queue耗绿。

通過增加consumer實(shí)例去分?jǐn)俼ueue的消費(fèi)，可以起到水平擴(kuò)展的消費(fèi)能力的作用砾隅。而有實(shí)例下線的時(shí)候误阻，會重新觸發(fā)負(fù)載均衡，這時(shí)候原來分配到的queue將分配到其他實(shí)例上繼續(xù)消費(fèi)晴埂。

2）廣播模式

由于廣播模式下要求一條消息需要投遞到一個(gè)消費(fèi)組下面所有的消費(fèi)者實(shí)例堕绩，所以也就沒有消息被分?jǐn)傁M(fèi)的說法。

在實(shí)現(xiàn)上邑时，其中一個(gè)不同就是在consumer分配queue的時(shí)候奴紧，所有consumer都分到所有的queue。

每個(gè)consumer實(shí)例分配每個(gè)consume queue

4 消息重試

4.1 順序消息重試

對于順序消息，當(dāng)消費(fèi)者消費(fèi)消息失敗后，消息隊(duì)列 RocketMQ 會自動不斷進(jìn)行消息重試（每次間隔時(shí)間為 1 秒）晕鹊，這時(shí)，應(yīng)用會出現(xiàn)消息消費(fèi)被阻塞的情況沫浆。因此，在使用順序消息時(shí)滚秩，務(wù)必保證應(yīng)用能夠及時(shí)監(jiān)控并處理消費(fèi)失敗的情況专执，避免阻塞現(xiàn)象的發(fā)生。

4.2 無序消息的重試

對于無序消息（普通郁油、定時(shí)本股、延時(shí)攀痊、事務(wù)消息），當(dāng)消費(fèi)者消費(fèi)消息失敗時(shí)拄显，您可以通過設(shè)置返回狀態(tài)達(dá)到消息重試的結(jié)果苟径。

無序消息的重試只針對 集群消費(fèi) 方式生效；廣播方式 不提供 失敗重試特性躬审，即消費(fèi)失敗后棘街，失敗消息不再重試，繼續(xù)消費(fèi)新的消息承边。

1）重試次數(shù)

消息隊(duì)列 RocketMQ 默認(rèn)允許每條消息最多重試 16 次遭殉，每次重試的間隔時(shí)間如下：

如果消息重試 16 次后仍然失敗，消息將不再投遞博助。如果嚴(yán)格按照上述重試時(shí)間間隔計(jì)算恩沽，某條消息在一直消費(fèi)失敗的前提下，將會在接下來的 4 小時(shí) 46 分鐘之內(nèi)進(jìn)行 16 次重試翔始，超過這個(gè)時(shí)間范圍消息將不再重試投遞。

注意： 一條消息無論重試多少次里伯，這些重試消息的 Message ID 不會改變城瞎。

2）配置方式

消費(fèi)失敗后，重試配置方式

集群消費(fèi)方式下疾瓮，消息消費(fèi)失敗后期望消息重試脖镀，需要在消息監(jiān)聽器接口的實(shí)現(xiàn)中明確進(jìn)行配置（三種方式任選一種）：

• 返回 Action.ReconsumeLater （推薦）
• 返回 Null
• 拋出異常

public class MessageListenerImpl implements MessageListener {
    @Override
    public Action consume(Message message, ConsumeContext context) {
        //處理消息
        doConsumeMessage(message);
        //方式1：返回 Action.ReconsumeLater，消息將重試
        return Action.ReconsumeLater;
        //方式2：返回 null狼电，消息將重試
        return null;
        //方式3：直接拋出異常蜒灰， 消息將重試
        throw new RuntimeException("Consumer Message exceotion");
    }
}

消費(fèi)失敗后，不重試配置方式
集群消費(fèi)方式下肩碟，消息失敗后期望消息不重試强窖，需要捕獲消費(fèi)邏輯中可能拋出的異常，最終返回 Action.CommitMessage削祈，此后這條消息將不會再重試翅溺。

public class MessageListenerImpl implements MessageListener {
    @Override
    public Action consume(Message message, ConsumeContext context) {
        try {
            doConsumeMessage(message);
        } catch (Throwable e) {
            //捕獲消費(fèi)邏輯中的所有異常，并返回 Action.CommitMessage;
            return Action.CommitMessage;
        }
        //消息處理正常髓抑，直接返回 Action.CommitMessage;
        return Action.CommitMessage;
    }
}

自定義消息最大重試次數(shù)
消息隊(duì)列 RocketMQ 允許 Consumer 啟動的時(shí)候設(shè)置最大重試次數(shù)咙崎，重試時(shí)間間隔將按照如下策略：

• 最大重試次數(shù)小于等于 16 次，則重試時(shí)間間隔同上表描述吨拍。
• 最大重試次數(shù)大于 16 次褪猛，超過 16 次的重試時(shí)間間隔均為每次 2 小時(shí)。

Properties properties = new Properties();
//配置對應(yīng) Group ID 的最大消息重試次數(shù)為 20 次
properties.put(PropertyKeyConst.MaxReconsumeTimes,"20");
Consumer consumer =ONSFactory.createConsumer(properties);

ps:

• 消息最大重試次數(shù)的設(shè)置對相同 Group ID 下的所有 Consumer 實(shí)例有效羹饰。
• 如果只對相同 Group ID 下兩個(gè) Consumer 實(shí)例中的其中一個(gè)設(shè)置了 MaxReconsumeTimes伊滋，那么該配置對兩個(gè) Consumer 實(shí)例均生效碳却。
• 配置采用覆蓋的方式生效，即最后啟動的 Consumer 實(shí)例會覆蓋之前的啟動實(shí)例的配置

獲取消息重試次數(shù)
消費(fèi)者收到消息后新啼，可按照如下方式獲取消息的重試次數(shù)：

public class MessageListenerImpl implements MessageListener {
    @Override
    public Action consume(Message message, ConsumeContext context) {
        //獲取消息的重試次數(shù)
        System.out.println(message.getReconsumeTimes());
        return Action.CommitMessage;
    }
}

5 死信隊(duì)列

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

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

5.1 死信特性

死信消息具有以下特性

• 不會再被消費(fèi)者正常消費(fèi)荠察。
• 有效期與正常消息相同置蜀，均為 3 天，3 天后會被自動刪除悉盆。因此盯荤，請?jiān)谒佬畔a(chǎn)生后的 3 天內(nèi)及時(shí)處理。

死信隊(duì)列具有以下特性：

• 一個(gè)死信隊(duì)列對應(yīng)一個(gè) Group ID焕盟， 而不是對應(yīng)單個(gè)消費(fèi)者實(shí)例秋秤。
• 如果一個(gè) Group ID 未產(chǎn)生死信消息，消息隊(duì)列 RocketMQ 不會為其創(chuàng)建相應(yīng)的死信隊(duì)列脚翘。
• 一個(gè)死信隊(duì)列包含了對應(yīng) Group ID 產(chǎn)生的所有死信消息灼卢，不論該消息屬于哪個(gè) Topic。

5.2 查看死信信息

1. 在控制臺查詢出現(xiàn)死信隊(duì)列的主題信息

image.png

2. 在消息界面根據(jù)主題查詢死信消息

image.png

3. 選擇重新發(fā)送消息

一條消息進(jìn)入死信隊(duì)列来农，意味著某些因素導(dǎo)致消費(fèi)者無法正常消費(fèi)該消息鞋真，因此，通常需要您對其進(jìn)行特殊處理沃于。排查可疑因素并解決問題后灿巧，可以在消息隊(duì)列 RocketMQ 控制臺重新發(fā)送該消息，讓消費(fèi)者重新消費(fèi)一次揽涮。

6 消費(fèi)冪等

消息隊(duì)列 RocketMQ 消費(fèi)者在接收到消息以后抠藕，有必要根據(jù)業(yè)務(wù)上的唯一 Key 對消息做冪等處理的必要性。（防止重復(fù)消費(fèi)）

6.1 消費(fèi)冪等的必要性

在互聯(lián)網(wǎng)應(yīng)用中蒋困，尤其在網(wǎng)絡(luò)不穩(wěn)定的情況下盾似，消息隊(duì)列 RocketMQ 的消息有可能會出現(xiàn)重復(fù)，這個(gè)重復(fù)簡單可以概括為以下情況：

• 發(fā)送時(shí)消息重復(fù)

  當(dāng)一條消息已被成功發(fā)送到服務(wù)端并完成持久化，此時(shí)出現(xiàn)了網(wǎng)絡(luò)閃斷或者客戶端宕機(jī)零院，導(dǎo)致服務(wù)端對客戶端應(yīng)答失敗溉跃。 如果此時(shí)生產(chǎn)者意識到消息發(fā)送失敗并嘗試再次發(fā)送消息，消費(fèi)者后續(xù)會收到兩條內(nèi)容相同并且 Message ID 也相同的消息告抄。

• 投遞時(shí)消息重復(fù)

  消息消費(fèi)的場景下撰茎，消息已投遞到消費(fèi)者并完成業(yè)務(wù)處理，當(dāng)客戶端給服務(wù)端反饋應(yīng)答的時(shí)候網(wǎng)絡(luò)閃斷打洼。 為了保證消息至少被消費(fèi)一次龄糊，消息隊(duì)列 RocketMQ 的服務(wù)端將在網(wǎng)絡(luò)恢復(fù)后再次嘗試投遞之前已被處理過的消息，消費(fèi)者后續(xù)會收到兩條內(nèi)容相同并且 Message ID 也相同的消息募疮。

• 負(fù)載均衡時(shí)消息重復(fù)（包括但不限于網(wǎng)絡(luò)抖動炫惩、Broker 重啟以及訂閱方應(yīng)用重啟）

  當(dāng)消息隊(duì)列 RocketMQ 的 Broker 或客戶端重啟、擴(kuò)容或縮容時(shí)阿浓，會觸發(fā) Rebalance他嚷，此時(shí)消費(fèi)者可能會收到重復(fù)消息。

6.2 處理方式

因?yàn)?Message ID 有可能出現(xiàn)沖突（重復(fù)）的情況芭毙，所以真正安全的冪等處理筋蓖，不建議以 Message ID 作為處理依據(jù)。 最好的方式是以業(yè)務(wù)唯一標(biāo)識作為冪等處理的關(guān)鍵依據(jù)退敦，而業(yè)務(wù)的唯一標(biāo)識可以通過消息 Key 進(jìn)行設(shè)置：

Message message = new Message();
message.setKey("ORDERID_100");
SendResult sendResult = producer.send(message);

訂閱方收到消息時(shí)可以根據(jù)消息的 Key 進(jìn)行冪等處理：

consumer.subscribe("ons_test", "*", new MessageListener() {
    public Action consume(Message message, ConsumeContext context) {
        String key = message.getKey()
        // 根據(jù)業(yè)務(wù)唯一標(biāo)識的 key 做冪等處理
    }
});