RocketMQ是是如何管理消費進度的？又是如何保證訊息成功消費的?-有解無憂

RocketMQ消費者保障

訊息確認機制

consumer的每個實體是靠佇列分配來決定如何消費訊息的，那么消費進度具體是如何管理的，又是如何保證訊息成功消費的?（RocketMQ有保證訊息肯定消費成功的特性,失敗則重試）

什么是ACK

訊息確認機制

在實際使用RocketMQ的時候我們并不能保證每次發送的訊息都剛好能被消費者一次性正常消費成功，可能會存在需要多次消費才能成功或者一直消費失敗的情況，那作為發送者該做如何處理呢？

為了保證資料不被丟失，RocketMQ支持訊息確認機制，即ack，發送者為了保證訊息肯定消費成功，只有使用方明確表示消費成功，RocketMQ才會認為訊息消費成功，中途斷電，拋出例外等都不會認為成功——即都會重新投遞，

保證資料能被正確處理而不僅僅是被Consumer收到，我們就不能采用no-ack或者auto-ack，我們需要手動ack(manual-ack)，在資料處理完成后手動發送ack，這個時候Server才將Message洗掉，

RocketMQ ACK

由于以上作業所有的機制都實作在PushConsumer中，所以本文的原理均只適用于RocketMQ中的PushConsumer即Java客戶端中的DefaultPushConsumer，若使用了PullConsumer模式，類似的作業如何ack，如何保證消費等均需要使用方自己實作，

注：廣播消費和集群消費的處理有部磁區別，以下均特指集群消費（CLSUTER），廣播（BROADCASTING）下部分可能不適用，

保證消費成功

PushConsumer為了保證訊息肯定消費成功，只有使用方明確表示消費成功，RocketMQ才會認為訊息消費成功，中途斷電，拋出例外等都不會認為成功——即都會重新投遞，

代碼示例

消費的時候，我們需要注入一個消費回呼，具體sample代碼如下：

COPYconsumer.registerMessageListener(new MessageListenerConcurrently() {
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
            System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs);
            execute();//執行真正消費
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    });

業務實作消費回呼的時候，當且僅當此回呼函式回傳ConsumeConcurrentlyStatus.CONSUME_SUCCESS，RocketMQ才會認為這批訊息（默認是1條）是消費完成的，

如果這時候訊息消費失敗，例如資料庫例外，余額不足扣款失敗等一切業務認為訊息需要重試的場景，只要回傳ConsumeConcurrentlyStatus.RECONSUME_LATER，RocketMQ就會認為這批訊息消費失敗了，

為了保證訊息是肯定被至少消費成功一次，RocketMQ會把這批訊息重發回Broker（topic不是原topic而是這個消費租的RETRY topic），在延遲的某個時間點（默認是10秒，業務可設定）后，再次投遞到這個ConsumerGroup，而如果一直這樣重復消費都持續失敗到一定次數（默認16次），就會投遞到DLQ死信佇列，應用可以監控死信佇列來做人工干預，

ACK進度保存

啟動的時候從哪里消費

當新實體啟動的時候，PushConsumer會拿到本消費組broker已經記錄好的消費進度（consumer offset），按照這個進度發起自己的第一次Pull請求，

如果這個消費進度在Broker并沒有存盤起來，證明這個是一個全新的消費組，這時候客戶端有幾個策略可以選擇：

COPYCONSUME_FROM_LAST_OFFSET //默認策略，從該佇列最尾開始消費，即跳過歷史訊息
CONSUME_FROM_FIRST_OFFSET //從佇列最開始開始消費，即歷史訊息（還儲存在broker的）全部消費一遍
CONSUME_FROM_TIMESTAMP//從某個時間點開始消費，和setConsumeTimestamp()配合使用，默認是半個小時以前

所以，社區中經常有人問：“為什么我設了CONSUME_FROM_LAST_OFFSET，歷史的訊息還是被消費了”？原因就在于只有全新的消費組才會使用到這些策略，老的消費組都是按已經存盤過的消費進度繼續消費，

對于老消費組想跳過歷史訊息需要自身做過濾，或者使用先修改消費進度

訊息ACK消費進度

RocketMQ是以consumer group+queue為單位是管理消費進度的，以一個consumer offset標記這個這個消費組在這條queue上的消費進度，

如果某已存在的消費組出現了新消費實體的時候，依靠這個組的消費進度，就可以判斷第一次是從哪里開始拉取的，每次訊息成功后，本地的消費進度會被更新，然后由定時器定時同步到broker，以此持久化消費進度，

但是每次記錄消費進度的時候，只會把一批訊息中最小的offset值為消費進度值，如下圖：

message ack

這鐘方式和傳統的一條message單獨ack的方式有本質的區別，性能上提升的同時，會帶來一個潛在的重復問題——由于消費進度只是記錄了一個下標，就可能出現拉取了100條訊息如 2101-2200的訊息，后面99條都消費結束了，只有2101消費一直沒有結束的情況，

在這種情況下，RocketMQ為了保證訊息肯定被消費成功，消費進度職能維持在2101，直到2101也消費結束了，本地的消費進度才能標記2200消費結束了（注：consumerOffset=2201），

重復消費

在這種設計下，就有消費大量重復的風險，如2101在還沒有消費完成的時候消費實體突然退出（機器斷電，或者被kill），這條queue的消費進度還是維持在2101，當queue重新分配給新的實體的時候，新的實體從broker上拿到的消費進度還是維持在2101，這時候就會又從2101開始消費，2102-2200這批訊息實際上已經被消費過還是會投遞一次，

對于這個場景，RocketMQ暫時無能為力，所以業務必須要保證訊息消費的冪等性，這也是RocketMQ官方多次強調的態度，

實際上，從原始碼的角度上看，RocketMQ可能是考慮過這個問題的，截止到3.2.6的版本的原始碼中，可以看到為了緩解這個問題的影響面，DefaultMQPushConsumer中有個配置consumeConcurrentlyMaxSpan

COPY/**
 * Concurrently max span offset.it has no effect on sequential consumption
 */
private int consumeConcurrentlyMaxSpan = 2000;

這個值默認是2000，當RocketMQ發現本地快取的訊息的最大值-最小值差距大于這個值（2000）的時候，會觸發流控——也就是說如果頭尾都卡住了部分訊息，達到了這個閾值就不再拉取訊息，

但作用實際很有限，像剛剛這個例子，2101的消費是死回圈，其他消費非常正常的話，是無能為力的，一旦退出，在不人工干預的情況下，2101后所有訊息全部重復!

Ack卡進度解決方案

實際上對于卡住進度的場景，可以選擇棄車保帥的方案：把訊息卡住那些訊息，先ack掉，讓進度前移，但要保證這條訊息不會因此丟失，ack之前要把訊息sendBack回去，這樣這條卡住的訊息就會必然重復，但會解決潛在的大量重復的場景，這也是我們公司自己定制的解決方案，

部分原始碼如下：

COPYclass ConsumeRequestWithUnAck implements Runnable {
    final ConsumeRequest consumeRequest;
    final long resendAfterIfStillUnAck;//n毫秒沒有消費完，就重發
 
    ConsumeRequestWithUnAck(ConsumeRequest consumeRequest,long resendAfterIfStillUnAck) {
        this.consumeRequest = consumeRequest;
        this.resendAfterIfStillUnAck = resendAfterIfStillUnAck;
    }
 
    @Override
    public void run() {
        //每次消費前，計劃延時任務，超時則ack并重發
        final WeakReference<ConsumeRequest> crReff = new WeakReference<>(this.consumeRequest);
        ScheduledFuture scheduledFuture=null;
        if(!ConsumeDispatcher.this.ackAndResendScheduler.isShutdown()) {
            scheduledFuture= ConsumeDispatcher.this.ackAndResendScheduler.schedule(new ConsumeTooLongChecker(crReff),resendAfterIfStillUnAck,TimeUnit.MILLISECONDS);
        }
        try{
            this.consumeRequest.run();//正常執行并更新offset
        }
        finally {
            if (scheduledFuture != null) scheduledFuture.cancel(false);//消費結束后,取消任務
        }
    }
 
}

定義了一個裝飾器，把原來的ConsumeRequest物件包了一層，
裝飾器中，每條訊息消費前都會調度一個調度器，定時觸發，觸發的時候如果發現訊息還存在，就執行sendback并ack的操作，

后來RocketMQ顯然也發現了這個問題，RocketMQ在3.5.8之后也是采用這樣的方案去解決這個問題，只是實作方式上有所不同（事實上我認為RocketMQ的方案還不夠完善）
在pushConsumer中有一個consumeTimeout欄位（默認15分鐘），用于設定最大的消費超時時間，消費前會記錄一個消費的開始時間，后面用于比對，
消費者啟動的時候，會定期掃描所有消費的訊息，達到這個timeout的那些訊息，就會觸發sendBack并ack的操作，這里掃描的間隔也是consumeTimeout（單位分鐘）的間隔，

核心原始碼如下：

COPY//ConsumeMessageConcurrentlyService.java
public void start() {
    this.CleanExpireMsgExecutors.scheduleAtFixedRate(new Runnable() {
 
        @Override
        public void run() {
            cleanExpireMsg();
        }
 
    }, this.defaultMQPushConsumer.getConsumeTimeout(), this.defaultMQPushConsumer.getConsumeTimeout(), TimeUnit.MINUTES);
}
//ConsumeMessageConcurrentlyService.java
private void cleanExpireMsg() {
    Iterator<Map.Entry<MessageQueue, ProcessQueue>> it =
            this.defaultMQPushConsumerImpl.getRebalanceImpl().getProcessQueueTable().entrySet().iterator();
    while (it.hasNext()) {
        Map.Entry<MessageQueue, ProcessQueue> next = it.next();
        ProcessQueue pq = next.getValue();
        pq.cleanExpiredMsg(this.defaultMQPushConsumer);
    }
}
 
//ProcessQueue.java
public void cleanExpiredMsg(DefaultMQPushConsumer pushConsumer) {
    if (pushConsumer.getDefaultMQPushConsumerImpl().isConsumeOrderly()) {
        return;
    }
 
    int loop = msgTreeMap.size() < 16 ? msgTreeMap.size() : 16;
    for (int i = 0; i < loop; i++) {
        MessageExt msg = null;
        try {
            this.lockTreeMap.readLock().lockInterruptibly();
            try {
                if (!msgTreeMap.isEmpty() && System.currentTimeMillis() - Long.parseLong(MessageAccessor.getConsumeStartTimeStamp(msgTreeMap.firstEntry().getValue())) > pushConsumer.getConsumeTimeout() * 60 * 1000) {
                    msg = msgTreeMap.firstEntry().getValue();
                } else {
 
                    break;
                }
            } finally {
                this.lockTreeMap.readLock().unlock();
            }
        } catch (InterruptedException e) {
            log.error("getExpiredMsg exception", e);
        }
 
        try {
 
            pushConsumer.sendMessageBack(msg, 3);
            log.info("send expire msg back. topic={}, msgId={}, storeHost={}, queueId={}, queueOffset={}", msg.getTopic(), msg.getMsgId(), msg.getStoreHost(), msg.getQueueId(), msg.getQueueOffset());
            try {
                this.lockTreeMap.writeLock().lockInterruptibly();
                try {
                    if (!msgTreeMap.isEmpty() && msg.getQueueOffset() == msgTreeMap.firstKey()) {
                        try {
                            msgTreeMap.remove(msgTreeMap.firstKey());
                        } catch (Exception e) {
                            log.error("send expired msg exception", e);
                        }
                    }
                } finally {
                    this.lockTreeMap.writeLock().unlock();
                }
            } catch (InterruptedException e) {
                log.error("getExpiredMsg exception", e);
            }
        } catch (Exception e) {
            log.error("send expired msg exception", e);
        }
    }
}

通過這個邏輯對比我定制的時間，可以看出有幾個不太完善的問題：

消費timeout的時間非常不精確，由于掃描的間隔是15分鐘，所以實際上觸發的時候，訊息是有可能卡住了接近30分鐘（15*2）才被清理，
由于定時器一啟動就開始調度了，中途這個consumeTimeout再更新也不會生效，

訊息重試

順序訊息的重試

對于順序訊息，當消費者消費訊息失敗后，訊息佇列RocketMQ版會自動不斷地進行訊息重試（每次間隔時間為1秒），這時，應用會出現訊息消費被阻塞的情況，因此，建議您使用順序訊息時，務必保證應用能夠及時監控并處理消費失敗的情況，避免阻塞現象的發生，

無序訊息的重試

對于無序訊息（普通、定時、延時、事務訊息），當消費者消費訊息失敗時，您可以通過設定回傳狀態達到訊息重試的結果，

無序訊息的重試只針對集群消費方式生效；廣播方式不提供失敗重試特性，即消費失敗后，失敗訊息不再重試，繼續消費新的訊息，

注意以下內容都只針對無序訊息生效，

重試次數

訊息佇列RocketMQ版默認允許每條訊息最多重試16次，每次重試的間隔時間如下，

第幾次重試	與上次重試的間隔時間	第幾次重試	與上次重試的間隔時間
1	10秒	9	7分鐘
2	30秒	10	8分鐘
3	1分鐘	11	9分鐘
4	2分鐘	12	10分鐘
5	3分鐘	13	20分鐘
6	4分鐘	14	30分鐘
7	5分鐘	15	1小時
8	6分鐘	16	2小時

如果訊息重試16次后仍然失敗，訊息將不再投遞，如果嚴格按照上述重試時間間隔計算，某條訊息在一直消費失敗的前提下，將會在接下來的4小時46分鐘之內進行16次重試，超過這個時間范圍訊息將不再重試投遞，

注意一條訊息無論重試多少次，這些重試訊息的Message ID不會改變，

和生產端重試區別

消費者和生產者的重試還是有區別的，主要有兩點

默認重試次數：Product默認是2次，而Consumer默認是16次，
重試時間間隔：Product是立刻重試，而Consumer是有一定時間間隔的，它照1S,5S,10S,30S,1M,2M····2H進行重試，
Product在異步情況重試失效，而對于Consumer在廣播情況下重試失效，

配置方式

重試配置方式

消費失敗后，重試配置方式，集群消費方式下，訊息消費失敗后期望訊息重試，需要在訊息監聽器介面的實作中明確進行配置（三種方式任選一種）：

方式1：回傳RECONSUME_LATER（推薦）
方式2：回傳Null
方式3：拋出例外

示例代碼

COPY//注冊訊息監聽器
consumer.registerMessageListener(new MessageListenerConcurrently() {
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
        //訊息處理邏輯拋出例外，訊息將重試，
        doConsumeMessage(list);
        //方式1：回傳ConsumeConcurrentlyStatus.RECONSUME_LATER，訊息將重試，
        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        //方式2：回傳null，訊息將重試，
        //  return null;
        //方式3：直接拋出例外，訊息將重試，
        // throw new RuntimeException("Consumer Message exception");
    }
});

無需重試的配置方式

集群消費方式下，訊息失敗后期望訊息不重試，需要捕獲消費邏輯中可能拋出的例外，最侄訓傳Action.CommitMessage，此后這條訊息將不會再重試，

示例代碼

COPY//注冊訊息監聽器
consumer.registerMessageListener(new MessageListenerConcurrently() {
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
        //訊息處理邏輯拋出例外，訊息將重試，
        try {
            doConsumeMessage(list);
        }catch (Exception e){
            //捕獲消費邏輯中的所有例外，并回傳Action.CommitMessage;
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
        //業務方正常消費
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
});

獲取訊息重試次數

消費者收到訊息后，可按照以下方式獲取訊息的重試次數：

COPY//注冊訊息監聽器
consumer.registerMessageListener(new MessageListenerConcurrently() {
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
        doConsumeMessage(list);
        //獲取訊息重試次數
        int retryTimes = list.get(0).getReconsumeTimes();
        //業務方正常消費
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
});

效果演示

消費端只發送一條訊息進行測驗重試

消費端主動拒絕

演示代碼

COPYpublic class RetryConsumer {
    public static void main(String[] args) throws Exception {
        //創建一個訊息消費者，并設定一個訊息消費者組
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("rocket_test_consumer_group");
        //指定 NameServer 地址
        consumer.setNamesrvAddr("127.0.0.1:9876");
        //設定 Consumer 第一次啟動時從佇列頭部開始消費還是佇列尾部開始消費
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        //訂閱指定 Topic 下的所有訊息
        consumer.subscribe("topicTest", "*");

        //注冊訊息監聽器
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
                if (list != null) {
                    for (MessageExt ext : list) {
                        //獲取訊息重試次數
                        int retryTimes = ext.getReconsumeTimes();
                        try {
                            String message = new String(ext.getBody(), RemotingHelper.DEFAULT_CHARSET);
                            System.out.println("Consumer-執行緒名稱=[" + Thread.currentThread().getId() + "],訊息重試次數:[" + retryTimes + "],接收時間:[" + new Date().getTime() + "],訊息=[" + message + "]");
                        } catch (UnsupportedEncodingException e) {
                            e.printStackTrace();
                        }
                    }
                }
                //業務主動拒絕訊息
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        });

        // 消費者物件在使用之前必須要呼叫 start 初始化
        consumer.start();
        System.out.println("訊息消費者已啟動");
    }
}

重試訊息

消費端回傳 ConsumeConcurrentlyStatus.RECONSUME_LATER，消費端就會不斷的重試，

COPYConsumer-執行緒名稱=[35],訊息重試次數:[0],接收時間:[1608117780264],訊息=[Hello Java demo RocketMQ ]
Consumer-執行緒名稱=[36],訊息重試次數:[1],接收時間:[1608117790840],訊息=[Hello Java demo RocketMQ ]
Consumer-線程名稱=[37],訊息重試次數:[2],接收時間:[1608117820876],訊息=[Hello Java demo RocketMQ ]

例外重試

演示代碼

這里的代碼意思很明顯: 主動拋出一個例外，然后如果超過3次，那么就不繼續重試下去，而是將該條記錄保存到資料庫由人工來兜底，

COPYpublic class RetryConsumer {
    public static void main(String[] args) throws Exception {
        //創建一個訊息消費者，并設定一個訊息消費者組
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("rocket_test_consumer_group");
        //指定 NameServer 地址
        consumer.setNamesrvAddr("127.0.0.1:9876");
        //設定 Consumer 第一次啟動時從佇列頭部開始消費還是佇列尾部開始消費
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        //訂閱指定 Topic 下的所有訊息
        consumer.subscribe("topicTest", "*");

        //注冊訊息監聽器
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
                if (list != null) {
                    for (MessageExt ext : list) {
                        //獲取訊息重試次數
                        int retryTimes = ext.getReconsumeTimes();
                        try {
                            String message = new String(ext.getBody(), RemotingHelper.DEFAULT_CHARSET);
                            System.out.println("Consumer-執行緒名稱=[" + Thread.currentThread().getId() + "],訊息重試次數:[" + retryTimes + "],接收時間:[" + new Date().getTime() + "],訊息=[" + message + "]");
                            //這里設定重試大于3次 那么通過保存資料庫 人工來兜底
                            if (retryTimes >= 2) {
                                System.out.println("該訊息已經重試3次,保存資料庫，topic=[" + ext.getTags() + "],keys=[" + ext.getKeys() + "],msg=[" + message + "]");
                                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                            }
                        } catch (UnsupportedEncodingException e) {
                            e.printStackTrace();
                        }
                    }
                }
                //主動拋出例外
                throw new RuntimeException("=======這里出錯了============");
                //業務方正常消費
                //return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 消費者物件在使用之前必須要呼叫 start 初始化
        consumer.start();
        System.out.println("訊息消費者已啟動");
    }
}

重試訊息

COPYConsumer-執行緒名稱=[36],訊息重試次數:[0],接收時間:[1608118304600],訊息=[Hello Java demo RocketMQ ]
Consumer-執行緒名稱=[37],訊息重試次數:[1],接收時間:[1608118315165],訊息=[Hello Java demo RocketMQ ]
Consumer-執行緒名稱=[38],訊息重試次數:[2],接收時間:[1608118345191],訊息=[Hello Java demo RocketMQ ]
該訊息已經重試3次,保存資料庫，topic=[TagA],keys=[null],msg=[Hello Java demo RocketMQ ]

超時重試

這里的超時例外并非真正意義上的超時，它指的是指獲取訊息后，因為某種原因沒有給RocketMQ回傳消費的狀態，即沒有return ConsumeConcurrentlyStatus.CONSUME_SUCCESS 或 return ConsumeConcurrentlyStatus.RECONSUME_LATER

那么 RocketMQ會認為該訊息沒有發送，會一直發送，因為它會認為該訊息根本就沒有發送給消費者,所以肯定沒消費，

演示代碼

COPYpublic class RetryConsumer {
    public static void main(String[] args) throws Exception {
        //創建一個訊息消費者，并設定一個訊息消費者組
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("rocket_test_consumer_group");
        //指定 NameServer 地址
        consumer.setNamesrvAddr("127.0.0.1:9876");
        //設定 Consumer 第一次啟動時從佇列頭部開始消費還是佇列尾部開始消費
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        //訂閱指定 Topic 下的所有訊息
        consumer.subscribe("topicTest", "*");

        //注冊訊息監聽器
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
                if (list != null) {
                    for (MessageExt ext : list) {
                        //獲取訊息重試次數
                        int retryTimes = ext.getReconsumeTimes();
                        try {
                            String message = new String(ext.getBody(), RemotingHelper.DEFAULT_CHARSET);
                            System.out.println("Consumer-執行緒名稱=[" + Thread.currentThread().getId() + "],訊息重試次數:[" + retryTimes + "],接收時間:[" + new Date().getTime() + "],訊息=[" + message + "]");
                        } catch (UnsupportedEncodingException e) {
                            e.printStackTrace();
                        }
                    }
                }
                //這里睡眠60秒
                try {
                    TimeUnit.SECONDS.sleep(60);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("休眠60秒 看還能不能走到這里...");
                //  業務方正常消費
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 消費者物件在使用之前必須要呼叫 start 初始化
        consumer.start();
        System.out.println("訊息消費者已啟動");
    }
}

重試訊息

當獲得 當前消費重試次數為 = 0 后 , 關掉該行程，再重新啟動該行程，那么依然能夠獲取該條訊息

COPYConsumer-執行緒名稱=[33],訊息重試次數:[0],接收時間:[1608118652598],訊息=[Hello Java demo RocketMQ ]

重啟消費者

COPYConsumer-執行緒名稱=[28],訊息重試次數:[0],接收時間:[1608118683304],訊息=[Hello Java demo RocketMQ ]
休眠60秒 看還能不能走到這里...

重試訊息的處理

一般情況下我們在實際生產中是不需要重試16次，這樣既浪費時間又浪費性能，理論上當嘗試重復次數達到我們想要的結果時如果還是消費失敗，那么我們需要將對應的訊息進行記錄，并且結束重復嘗試，

COPYpublic class RetryConsumer {
    public static void main(String[] args) throws Exception {
        //創建一個訊息消費者，并設定一個訊息消費者組
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("rocket_test_consumer_group");
        //指定 NameServer 地址
        consumer.setNamesrvAddr("127.0.0.1:9876");
        //設定 Consumer 第一次啟動時從佇列頭部開始消費還是佇列尾部開始消費
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        //訂閱指定 Topic 下的所有訊息
        consumer.subscribe("topicTest", "*");

        //注冊訊息監聽器
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
                if (list != null) {
                    for (MessageExt ext : list) {
                        //獲取訊息重試次數
                        int retryTimes = ext.getReconsumeTimes();
                        try {
                            String message = new String(ext.getBody(), RemotingHelper.DEFAULT_CHARSET);
                            System.out.println("Consumer-執行緒名稱=[" + Thread.currentThread().getId() + "],訊息重試次數:[" + retryTimes + "],接收時間:[" + new Date().getTime() + "],訊息=[" + message + "]");
                            //這里設定重試大于3次 那么通過保存資料庫 人工來兜底
                            if (retryTimes >= 2) {
                                System.out.println("該訊息已經重試3次,保存資料庫，topic=[" + ext.getTags() + "],keys=[" + ext.getKeys() + "],msg=[" + message + "]");
                                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                            }
                        } catch (UnsupportedEncodingException e) {
                            e.printStackTrace();
                        }
                    }
                }
                //主動拋出例外
                throw new RuntimeException("=======這里出錯了============");
                //業務方正常消費
                //return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 消費者物件在使用之前必須要呼叫 start 初始化
        consumer.start();
        System.out.println("訊息消費者已啟動");
    }
}

所以任何例外都要捕獲回傳ConsumeConcurrentlyStatus.RECONSUME_LATER,rocketmq會放到重試佇列,這個重試TOPIC的名字是%RETRY%+consumergroup的名字,如下圖：

注意點

如果業務的回呼沒有處理好而拋出例外，會認為是消費失敗當ConsumeConcurrentlyStatus.RECONSUME_LATER處理，
當使用順序消費的回呼MessageListenerOrderly時，由于順序消費是要前者消費成功才能繼續消費，所以沒有ConsumeConcurrentlyStatus.RECONSUME_LATER的這個狀態，只有ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT來暫停佇列的其余消費，直到原訊息不斷重試成功為止才能繼續消費，

RocketMQ重試流程

重試佇列與死信佇列

在介紹RocketMQ的消費重試機制之前，需要先來說下“重試佇列”和“死信佇列”兩個概念，

重試佇列

如果Consumer端因為各種型別例外導致本次消費失敗，為防止該訊息丟失而需要將其重新回發給Broker端保存，保存這種因為例外無法正常消費而回發給MQ的訊息佇列稱之為重試佇列，RocketMQ會為每個消費組都設定一個Topic名稱為“%RETRY%+consumerGroup”的重試佇列（這里需要注意的是，這個Topic的重試佇列是針對消費組，而不是針對每個Topic設定的），用于暫時保存因為各種例外而導致Consumer端無法消費的訊息，考慮到例外恢復起來需要一些時間，會為重試佇列設定多個重試級別，每個重試級別都有與之對應的重新投遞延時，重試次數越多投遞延時就越大，RocketMQ對于重試訊息的處理是先保存至Topic名稱為“SCHEDULE_TOPIC_XXXX”的延遲佇列中，后臺定時任務按照對應的時間進行Delay后重新保存至“%RETRY%+consumerGroup”的重試佇列中，

死信佇列

由于有些原因導致Consumer端長時間的無法正常消費從Broker端Pull過來的業務訊息，為了確保訊息不會被無故的丟棄，那么超過配置的“最大重試消費次數”后就會移入到這個死信佇列中，在RocketMQ中，SubscriptionGroupConfig配置常量默認地設定了兩個引數，一個是retryQueueNums為1（重試佇列數量為1個），另外一個是retryMaxTimes為16（最大重試消費的次數為16次），Broker端通過校驗判斷，如果超過了最大重試消費次數則會將訊息移至這里所說的死信佇列，這里，RocketMQ會為每個消費組都設定一個Topic命名為“%DLQ%+consumerGroup”的死信佇列，一般在實際應用中，移入至死信佇列的訊息，需要人工干預處理；

Consumer端回發訊息至Broker端

在業務工程中的Consumer端（Push消費模式下），如果訊息能夠正常消費需要在注冊的訊息監聽回呼方法中回傳CONSUME_SUCCESS的消費狀態，否則因為各類例外消費失敗則回傳RECONSUME_LATER的消費狀態，消費狀態的列舉型別如下所示：

COPYpublic enum ConsumeConcurrentlyStatus {
    //業務方消費成功
    CONSUME_SUCCESS,
    //業務方消費失敗，之后進行重新嘗試消費
    RECONSUME_LATER;
}

如果業務工程對訊息消費失敗了，那么則會拋出例外并且回傳這里的RECONSUME_LATER狀態，這里，在消費訊息的服務執行緒—consumeMessageService中，將封裝好的訊息消費任務ConsumeRequest提交至執行緒池—consumeExecutor異步執行，從訊息消費任務ConsumeRequest的run()方法中會執行業務工程中注冊的訊息監聽回呼方法，并在processConsumeResult方法中根據業務工程回傳的狀態（CONSUME_SUCCESS或者RECONSUME_LATER）進行判斷和做對應的處理，

CONSUME_SUCCESS

業務方正常消費

正常情況下，設定ackIndex的值為consumeRequest.getMsgs().size() - 1，因此后面的遍歷consumeRequest.getMsgs()訊息集合條件不成立，不會呼叫回發消費失敗訊息至Broker端的方法—sendMessageBack(msg, context)，最后，更新消費的偏移量；

RECONSUME_LATER

業務方消費失敗

例外情況下，設定ackIndex的值為-1，這時就會進入到遍歷consumeRequest.getMsgs()訊息集合的for回圈中，執行回發訊息的方法—sendMessageBack(msg, context)，這里，首先會根據brokerName得到Broker端的地址資訊，然后通過網路通信的Remoting模塊發送RPC請求到指定的Broker上，如果上述程序失敗，則創建一條新的訊息重新發送給Broker，此時新訊息的Topic為“%RETRY%+ConsumeGroupName”—重試佇列的主題，其中，在MQClientAPIImpl實體的consumerSendMessageBack()方法中封裝了ConsumerSendMsgBackRequestHeader的請求體，隨后完成回發消費失敗訊息的RPC通信請求（業務請求碼為：CONSUMER_SEND_MSG_BACK），倘若上面的回發訊息流程失敗，則會延遲5S后重新在Consumer端進行重新消費，與正常消費的情況一樣，在最后更新消費的偏移量；

Broker端對于回發訊息處理的主要流程

Broker端收到這條Consumer端回發過來的訊息后，通過業務請求碼（CONSUMER_SEND_MSG_BACK）匹配業務處理器—SendMessageProcessor來處理，在完成一系列的前置校驗（這里主要是“消費分組是否存在”、“檢查Broker是否有寫入權限”、“檢查重試佇列數是否大于0”等）后，嘗試獲取重試佇列的TopicConfig物件（如果是第一次無法獲取到，則呼叫createTopicInSendMessageBackMethod()方法進行創建），根據回發過來的訊息偏移量嘗試從commitlog日志檔案中查詢訊息內容，若不存在則回傳例外錯誤，

然后，設定重試佇列的Topic—“%RETRY%+consumerGroup”至MessageExt的擴展屬性“RETRY_TOPIC”中，并對根據延遲級別delayLevel和最大重試消費次數maxReconsumeTimes進行判斷，如果超過最大重試消費次數（默認16次），則會創建死信佇列的TopicConfig物件（用于后面將回發過來的訊息移入死信佇列），在構建完成需要落盤的MessageExtBrokerInner物件后，呼叫“commitLog.putMessage(msg)”方法做訊息持久化，這里，需要注意的是，在putMessage(msg)的方法里會使用“SCHEDULE_TOPIC_XXXX”和對應的延遲級別佇列Id分別替換MessageExtBrokerInner物件的Topic和QueueId屬性值，并將原來設定的重試佇列主題（“%RETRY%+consumerGroup”）的Topic和QueueId屬性值做一個備份分別存入擴展屬性properties的“REAL_TOPIC”和“REAL_QID”屬性中，看到這里也就大致明白了，回發給Broker端的消費失敗的訊息并非直接保存至重試佇列中，而是會先存至Topic為“SCHEDULE_TOPIC_XXXX”的定時延遲佇列中，

疑問：上面說了RocketMQ的重試佇列的Topic是“%RETRY%+consumerGroup”，為啥這里要保存至Topic是“SCHEDULE_TOPIC_XXXX”的這個延遲佇列中呢？

在原始碼中搜索下關鍵字—“SCHEDULE_TOPIC_XXXX”，會發現Broker端還存在著一個后臺服務執行緒—ScheduleMessageService（通過訊息存盤服務—DefaultMessageStore啟動），通過查看原始碼可以知道其中有一個DeliverDelayedMessageTimerTask定時任務執行緒會根據Topic（“SCHEDULE_TOPIC_XXXX”）與QueueId，先查到邏輯消費佇列ConsumeQueue，然后根據偏移量，找到ConsumeQueue中的記憶體映射物件，從commitlog日志中找到訊息物件MessageExt，并做一個訊息體的轉換（messageTimeup()方法，由定時延遲佇列訊息轉化為重試佇列的訊息），再次做持久化落盤，這時候才會真正的保存至重試佇列中，看到這里就可以解釋上面的疑問了，定時延遲佇列只是為了用于暫存的，然后延遲一段時間再將訊息移入至重試佇列中，RocketMQ設定不同的延時級別delayLevel，并且與定時延遲佇列相對應，具體原始碼如下：

COPY//省略
private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
/**
  * 定時延時訊息主題的佇列與延遲等級對應關系
  * @param delayLevel
  * @return
  */
public static int delayLevel2QueueId(final int delayLevel) {
    return delayLevel - 1;
}

Consumer端消費重試機制

每個Consumer實體在啟動的時候就默認訂閱了該消費組的重試佇列主題，DefaultMQPushConsumerImpl的copySubscription()方法中的相關代碼如下：

COPYprivate void copySubscription() throws MQClientException {
            //省略其他代碼...
            switch (this.defaultMQPushConsumer.getMessageModel()) {
                case BROADCASTING:
                    break;
                case CLUSTERING://如果訊息消費模式為集群模式，還需要為該消費組對應一個重試主題
                    final String retryTopic = MixAll.getRetryTopic(this.defaultMQPushConsumer.getConsumerGroup());
                    SubscriptionData subscriptionData = https://www.cnblogs.com/jiagooushi/p/FilterAPI.buildSubscriptionData(this.defaultMQPushConsumer.getConsumerGroup(),
                        retryTopic, SubscriptionData.SUB_ALL);
                    this.rebalanceImpl.getSubscriptionInner().put(retryTopic, subscriptionData);
                    break;
                default:
                    break;
            }
            //省略其他代碼...
      }

因此，這里也就清楚了，Consumer端會一直訂閱該重試佇列主題的訊息，向Broker端發送如下的拉取訊息的PullRequest請求，以嘗試重新再次消費重試佇列中積壓的訊息，

COPYPullRequest [consumerGroup=CID_JODIE_1, messageQueue=MessageQueue [topic=%RETRY%CID_JODIE_1, brokerName=HQSKCJJIDRRD6KC, queueId=0], nextOffset=51]

最后，給出一張RocketMQ訊息重試機制的框圖（ps：這里只是描述了訊息消費失敗后重試拉取的部分重要程序）：

本文由傳智教育博學谷狂野架構師教研團隊發布，

如果本文對您有幫助，歡迎關注和點贊；如果您有任何建議也可留言評論或私信，您的支持是我堅持創作的動力，

轉載請注明出處！

轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/549946.html

標籤：Java

上一篇：面向物件程式設計

下一篇：如何實作 Java SpringBoot 自動驗證入參資料的有效性