引言

本文主要介紹在使用 RocketMQ 時(shí)為什么需要重試與兜底機(jī)制，生產(chǎn)者與消費(fèi)者觸發(fā)重試的條件和具體行為，如何在 RocketMQ 中合理使用重試機(jī)制，幫助構(gòu)建彈性，高可用系統(tǒng)的最佳實(shí)踐。

RocketMQ 的重試機(jī)制包括三部分，分別是生產(chǎn)者重試，服務(wù)端內(nèi)部數(shù)據(jù)復(fù)制遇到非預(yù)期問題時(shí)重試，消費(fèi)者消費(fèi)重試。本文中僅討論生產(chǎn)者重試和消費(fèi)者消費(fèi)重試兩種面向用戶側(cè)的實(shí)現(xiàn)。

生產(chǎn)者發(fā)送重試

Cloud Native

RocketMQ 的生產(chǎn)者在發(fā)送消息到服務(wù)端時(shí)，可能會(huì)因?yàn)榫W(wǎng)絡(luò)問題，服務(wù)異常等原因?qū)е抡{(diào)用失敗，這時(shí)候應(yīng)該怎么辦？如何盡可能的保證消息不丟失呢？

1. 生產(chǎn)者重試次數(shù)

RocketMQ 在客戶端中內(nèi)置了請(qǐng)求重試邏輯，支持在初始化時(shí)配置消息發(fā)送最大重試次數(shù)（默認(rèn)為 2 次），失敗時(shí)會(huì)按照設(shè)置的重試次數(shù)重新發(fā)送。直到消息發(fā)送成功，或者達(dá)到最大重試次數(shù)時(shí)結(jié)束，并在最后一次失敗后返回調(diào)用錯(cuò)誤的響應(yīng)。對(duì)于同步發(fā)送和異步發(fā)送，均支持消息發(fā)送重試。

同步發(fā)送：調(diào)用線程會(huì)一直阻塞，直到某次重試成功或最終重試失?。ǚ祷劐e(cuò)誤碼或拋出異常）。

異步發(fā)送：調(diào)用線程不會(huì)阻塞，但調(diào)用結(jié)果會(huì)通過回調(diào)的形式，以異常事件或者成功事件返回。

2. 生產(chǎn)者重試間隔

在介紹生產(chǎn)者重試前，我們先來了解下流控的概念，流控一般是指服務(wù)端壓力過大，容量不足時(shí)服務(wù)端會(huì)限制客戶端收發(fā)消息的行為，是服務(wù)端自我保護(hù)的一種設(shè)計(jì)。RocketMQ 會(huì)根據(jù)當(dāng)前是否觸發(fā)了流控而采用不同的重試策略：

非流控錯(cuò)誤場(chǎng)景：其他觸發(fā)條件觸發(fā)重試后，均會(huì)立即進(jìn)行重試，無等待間隔。流控錯(cuò)誤場(chǎng)景：系統(tǒng)會(huì)按照預(yù)設(shè)的指數(shù)退避策略進(jìn)行延遲重試。

為什么要引入退避和隨機(jī)抖動(dòng)？

如果故障是由過載流控引起的，重試會(huì)增加服務(wù)端負(fù)載，導(dǎo)致情況進(jìn)一步惡化，因此客戶端在遇到流控時(shí)會(huì)在兩次嘗試之間等待一段時(shí)間。每次嘗試后的等待時(shí)間都呈指數(shù)級(jí)延長(zhǎng)。指數(shù)回退可能導(dǎo)致很長(zhǎng)的回退時(shí)間，因?yàn)橹笖?shù)函數(shù)增長(zhǎng)很快。指數(shù)退避算法通過以下參數(shù)控制重試行為，更多信息，請(qǐng)參見 connection-backoff.md。INITIAL_BACKOFF：第一次失敗重試前后需等待多久，默認(rèn)值：1 秒；

MULTIPLIER ：指數(shù)退避因子，即退避倍率，默認(rèn)值：1.6；

JITTER ：隨機(jī)抖動(dòng)因子，默認(rèn)值：0.2；

MAX_BACKOFF ：等待間隔時(shí)間上限，默認(rèn)值：120 秒；

MIN_CONNECT_TIMEOUT ：最短重試間隔，默認(rèn)值：20 秒。

ConnectWithBackoff（）

current_backoff = INITIAL_BACKOFF

current_deadline = now（） + INITIAL_BACKOFF

while （TryConnect（Max（current_deadline， now（） + MIN_CONNECT_TIMEOUT））！= SUCCESS）

SleepUntil（current_deadline）

current_backoff = Min（current_backoff * MULTIPLIER， MAX_BACKOFF）

current_deadline = now（） + current_backoff + UniformRandom（-JITTER * current_backoff， JITTER * current_backoff）

特別說明：對(duì)于事務(wù)消息，只會(huì)進(jìn)行透明重試（transparent retries），網(wǎng)絡(luò)超時(shí)或異常等場(chǎng)景不會(huì)進(jìn)行重試。

3. 重試帶來的副作用

不停的重試看起來很美好，但也是有副作用的，主要包括兩方面：消息重復(fù)，服務(wù)端壓力增大

遠(yuǎn)程調(diào)用的不確定性，因請(qǐng)求超時(shí)觸發(fā)消息發(fā)送重試流程，此時(shí)客戶端無法感知服務(wù)端的處理結(jié)果；客戶端進(jìn)行的消息發(fā)送重試可能會(huì)導(dǎo)致消費(fèi)方重復(fù)消費(fèi)，應(yīng)該按照用戶ID、業(yè)務(wù)主鍵等信息冪等處理消息。

較多的重試次數(shù)也會(huì)增大服務(wù)端的處理壓力。

4. 用戶的最佳實(shí)踐是什么

1）合理設(shè)置發(fā)送超時(shí)時(shí)間，發(fā)送的最大次數(shù)

發(fā)送的最大次數(shù)在初始化客戶端時(shí)配置在 ClientConfiguration；對(duì)于某些實(shí)時(shí)調(diào)用類場(chǎng)景，可能會(huì)導(dǎo)致消息發(fā)送請(qǐng)求鏈路被阻塞導(dǎo)致業(yè)務(wù)請(qǐng)求整體耗時(shí)高或耗時(shí)；需要合理評(píng)估每次調(diào)用請(qǐng)求的超時(shí)時(shí)間以及最大重試次數(shù)，避免影響全鏈路的耗時(shí)。2）如何保證發(fā)送消息不丟失由于分布式環(huán)境的復(fù)雜性，例如網(wǎng)絡(luò)不可達(dá)時(shí) RocketMQ 客戶端發(fā)送請(qǐng)求重試機(jī)制并不能保證消息發(fā)送一定成功。業(yè)務(wù)方需要捕獲異常，并做好冗余保護(hù)處理，常見的解決方案有兩種：

向調(diào)用方返回業(yè)務(wù)處理失?。?/span>

嘗試將失敗的消息存儲(chǔ)到數(shù)據(jù)庫(kù)，然后由后臺(tái)線程定時(shí)重試，保證業(yè)務(wù)邏輯的最終一致性。

3）關(guān)注流控異常導(dǎo)致無法重試觸發(fā)流控的根本原因是系統(tǒng)容量不足，如果因?yàn)橥话l(fā)原因觸發(fā)消息流控，且客戶端內(nèi)置的重試流程執(zhí)行失敗，

則建議執(zhí)行服務(wù)端擴(kuò)容，將請(qǐng)求調(diào)用臨時(shí)替換到其他系統(tǒng)進(jìn)行應(yīng)急處理。4）早期版本客戶端如何使用故障延遲機(jī)制進(jìn)行發(fā)送重試？對(duì)于 RocketMQ 4.x 和 3.x 以下客戶端開啟故障延遲機(jī)制可以用：

producer.setSendLatencyFaultEnable（true）

配置重試次數(shù)使用：

producer.setRetryTimesWhenSendFailed（）producer.setRetryTimesWhenSendAsyncFailed（）

消費(fèi)者消費(fèi)重試

Cloud Native

消息中間件做異步解耦時(shí)的一個(gè)典型問題是如果下游服務(wù)處理消息事件失敗，那應(yīng)該怎么做呢？RocketMQ 的消息確認(rèn)機(jī)制以及消費(fèi)重試策略可以幫助分析如下問題：

如何保證業(yè)務(wù)完整處理消息？

消費(fèi)重試策略可以在設(shè)計(jì)實(shí)現(xiàn)消費(fèi)者邏輯時(shí)保證每條消息處理的完整性，避免部分消息消費(fèi)異常導(dǎo)致業(yè)務(wù)狀態(tài)不一致。

業(yè)務(wù)應(yīng)用異常時(shí)處理中的消息狀態(tài)如何恢復(fù)？

當(dāng)系統(tǒng)出現(xiàn)異常（宕機(jī)故障）等場(chǎng)景時(shí)，處理中的消息狀態(tài)如何恢復(fù)，消費(fèi)重試具體行為是什么。

1. 什么是消費(fèi)重試？

什么時(shí)候認(rèn)為消費(fèi)失?。?/span>

消費(fèi)者在接收到消息后將調(diào)用用戶的消費(fèi)函數(shù)執(zhí)行業(yè)務(wù)邏輯。如果客戶端返回消費(fèi)失敗 ReconsumeLater，拋出非預(yù)期異常，或消息處理超時(shí)（包括在 PushConsumer 中排隊(duì)超時(shí)），只要服務(wù)端服務(wù)端一定時(shí)間內(nèi)沒收到響應(yīng)，將認(rèn)為消費(fèi)失敗。

消費(fèi)重試是什么？

消費(fèi)者在消費(fèi)某條消息失敗后，服務(wù)端會(huì)根據(jù)重試策略重新向客戶端投遞該消息。超過一次定數(shù)后若還未消費(fèi)成功，則該消息將不再繼續(xù)重試，直接被發(fā)送到死信隊(duì)列中；

重試過程狀態(tài)機(jī)：消息在重試流程中的狀態(tài)和變化邏輯；

重試間隔：上一次消費(fèi)失敗或超時(shí)后，下次重新嘗試消費(fèi)的間隔時(shí)間；

最大重試次數(shù)：消息可被重試消費(fèi)的最大次數(shù)。

2. 消息重試的場(chǎng)景

需要注意重試是應(yīng)對(duì)異常情況，給予程序再次消費(fèi)失敗消息的機(jī)會(huì)，不應(yīng)該被用作常態(tài)化的鏈路。

推薦使用場(chǎng)景：

業(yè)務(wù)處理失敗，失敗原因跟當(dāng)前的消息內(nèi)容相關(guān)，預(yù)期一段時(shí)間后可執(zhí)行成功；

是一個(gè)小概率事件，對(duì)于大批的消息只有很少量的失敗，后面的消息大概率會(huì)消費(fèi)成功，是非常態(tài)化的。

正例：消費(fèi)邏輯是扣減庫(kù)存，極少量商品因?yàn)闃酚^鎖版本沖突導(dǎo)致扣減失敗，重試一般立刻成功。錯(cuò)誤使用場(chǎng)景：

消費(fèi)處理邏輯中使用消費(fèi)失敗來做條件判斷的結(jié)果分流，是不合理的。

反例：訂單在數(shù)據(jù)庫(kù)中狀態(tài)已經(jīng)是已取消，此時(shí)如果收到發(fā)貨的消息，處理時(shí)不應(yīng)返回消費(fèi)失敗，而應(yīng)該返回成功并標(biāo)記不用發(fā)貨。

消費(fèi)處理中使用消費(fèi)失敗來做處理速率限流，是不合理的。

限流的目的是將超出流量的消息暫時(shí)堆積在隊(duì)列中達(dá)到削峰的作用，而不是讓消息進(jìn)入重試鏈路。

這種做法會(huì)讓消息反復(fù)在服務(wù)端和客戶端之間傳遞，增大了系統(tǒng)的開銷，主要包括以下方面：

RocketMQ 內(nèi)部重試涉及寫放大，每一次重試將生成新的重試消息，大量重試將帶來嚴(yán)重的 IO 壓力；

重試有復(fù)雜的退避邏輯，內(nèi)部實(shí)現(xiàn)為梯度定時(shí)器，該定時(shí)器本身不具備高吞吐的特性，大量重試將導(dǎo)致重試消息無法及時(shí)出隊(duì)。重試的間隔將不穩(wěn)定，將導(dǎo)致大量重試消息延后消費(fèi)，即削峰的周期被大幅度延長(zhǎng)。

3. 不要以重試替代限流

上述誤用的場(chǎng)景實(shí)際上是組合了限流和重試能力來進(jìn)行削峰，RocketMQ 推薦的削峰最佳手段為組合限流和堆積，業(yè)務(wù)以保護(hù)自身為前提，需要對(duì)消費(fèi)流量進(jìn)行限流，并利用 RocketMQ 提供的堆積能力將超出業(yè)務(wù)當(dāng)前處理的消息滯后消費(fèi)，以達(dá)到削峰的目的。下圖中超過處理能力的消息都應(yīng)該被堆積在服務(wù)端，而不是通過消費(fèi)失敗進(jìn)行重試。

如果不想依賴額外的產(chǎn)品/組件來完成該功能，也可以利用一些本地工具類，比如 Guava 的 RateLimiter 來完成單機(jī)限流。如下所示，聲明一個(gè) 50 QPS 的 RateLimiter，在消費(fèi)前以阻塞的方式 acquire 一個(gè)令牌，獲取到即處理消息，未獲取到阻塞。

4. PushConsumer 消費(fèi)重試策略

PushConsumer 消費(fèi)消息時(shí)，消息的幾個(gè)主要狀態(tài)如下：

Ready：已就緒狀態(tài)。消息在消息隊(duì)列RocketMQ版服務(wù)端已就緒，可以被消費(fèi)者消費(fèi)；

Inflight：處理中狀態(tài)。消息被消費(fèi)者客戶端獲取，處于消費(fèi)中還未返回消費(fèi)結(jié)果的狀態(tài)；

Commit：提交狀態(tài)。消費(fèi)成功的狀態(tài)，消費(fèi)者返回成功響應(yīng)即可結(jié)束消息的狀態(tài)機(jī)；

DLQ：死信狀態(tài)

消費(fèi)邏輯的最終兜底機(jī)制，若消息一直處理失敗并不斷進(jìn)行重試，直到超過最大重試次數(shù)還未成功，此時(shí)消息不會(huì)再重試。

該消息會(huì)被投遞至死信隊(duì)列。您可以通過消費(fèi)死信隊(duì)列的消息進(jìn)行業(yè)務(wù)恢復(fù)。

最大重試次數(shù)

PushConsumer 的最大重試次數(shù)由創(chuàng)建時(shí)決定。例如，最大重試次數(shù)為 3 次，則該消息最多可被投遞 4 次，1 次為原始消息，3 次為重試投遞次數(shù)。

重試間隔時(shí)間

無序消息（非順序消息）：重試間隔為階梯時(shí)間，具體時(shí)間如下：

說明：若重試次數(shù)超過 16 次，后面每次重試間隔都為 2 小時(shí)。

順序消息：重試間隔為固定時(shí)間，默認(rèn)為 3 秒。

5. SimpleConsumer 消費(fèi)重試策略

和 PushConsumer 消費(fèi)重試策略不同，SimpleConsumer 消費(fèi)者的重試間隔是預(yù)分配的，每次獲取消息消費(fèi)者會(huì)在調(diào)用 API 時(shí)設(shè)置一個(gè)不可見時(shí)間參數(shù) InvisibleDuration，即消息的最大處理時(shí)長(zhǎng)。若消息消費(fèi)失敗觸發(fā)重試，不需要設(shè)置下一次重試的時(shí)間間隔，直接復(fù)用不可見時(shí)間參數(shù)的取值。

由于不可見時(shí)間為預(yù)分配的，可能和實(shí)際業(yè)務(wù)中的消息處理時(shí)間差別較大，可以通過 API 接口修改不可見時(shí)間。例如，預(yù)設(shè)消息處理耗時(shí)最多 20 ms，但實(shí)際業(yè)務(wù)中 20 ms內(nèi)消息處理不完，可以修改消息不可見時(shí)間，延長(zhǎng)消息處理時(shí)間，避免消息觸發(fā)重試機(jī)制。修改消息不可見時(shí)間需要滿足以下條件：

消息處理未超時(shí)

消息處理未提交消費(fèi)狀態(tài)

如下圖所示，消息不可見時(shí)間修改后立即生效，即從調(diào)用 API 時(shí)刻開始，重新計(jì)算消息不可見時(shí)間。

最大重試次數(shù)

與 PushConsumer 相同。

消息重試間隔

消息重試間隔 = 不可見時(shí)間 － 消息實(shí)際處理時(shí)長(zhǎng)例如：消息不可見時(shí)間為 30 ms，實(shí)際消息處理用了 10 ms 就返回失敗響應(yīng)，則距下次消息重試還需要 20 ms，此時(shí)的消息重試間隔即為 20 ms；若直到 30 ms 消息還未處理完成且未返回結(jié)果，則消息超時(shí)，立即重試，此時(shí)重試間隔即為 0 ms。SimpleConsumer 的消費(fèi)重試間隔通過消息的不可見時(shí)間控制。

//消費(fèi)示例：使用SimpleConsumer消費(fèi)普通消息，主動(dòng)獲取消息處理并提交。
ClientServiceProvider provider1 = ClientServiceProvider.loadService();
String topic1 = "Your Topic";
FilterExpression filterExpression1 = new FilterExpression("Your Filter Tag", FilterExpressionType.TAG);


SimpleConsumer simpleConsumer = provider1.newSimpleConsumerBuilder()
        //設(shè)置消費(fèi)者分組。
        .setConsumerGroup("Your ConsumerGroup")
        //設(shè)置接入點(diǎn)。
        .setClientConfiguration(ClientConfiguration.newBuilder().setEndpoints("Your Endpoint").build())
        //設(shè)置預(yù)綁定的訂閱關(guān)系。
        .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
        .build();
List messageViewList = null;
try {
    //SimpleConsumer需要主動(dòng)獲取消息，并處理。
    messageViewList = simpleConsumer.receive(10, Duration.ofSeconds(30));
    messageViewList.forEach(messageView -> {
        System.out.println(messageView);
        //消費(fèi)處理完成后，需要主動(dòng)調(diào)用ACK提交消費(fèi)結(jié)果。
        //沒有ack會(huì)被認(rèn)為消費(fèi)失敗
        try {
            simpleConsumer.ack(messageView);
        } catch (ClientException e) {
            e.printStackTrace();
        }
    });
} catch (ClientException e) {
    //如果遇到系統(tǒng)流控等原因造成拉取失敗，需要重新發(fā)起獲取消息請(qǐng)求。
    e.printStackTrace();
}

案例：某產(chǎn)品使用消息隊(duì)列來發(fā)送解耦“視頻渲染”的業(yè)務(wù)邏輯，發(fā)送方發(fā)送任務(wù)編號(hào)，消費(fèi)方收到編號(hào)后處理任務(wù)。由于消費(fèi)方的業(yè)務(wù)邏輯耗時(shí)較長(zhǎng)，消費(fèi)者重新消費(fèi)到同一個(gè)任務(wù)時(shí)，該任務(wù)未完成，只能返回消費(fèi)失敗。在這種全新的 API 下，用戶可以調(diào)用可以通過修改不可見時(shí)間給消息續(xù)期，實(shí)現(xiàn)對(duì)單條消息狀態(tài)的精確控制。

simpleConsumer.changeInvisibleDuration（）;

simpleConsumer.changeInvisibleDurationAsync（）;

6. 功能約束與最佳實(shí)踐

設(shè)置消費(fèi)的最大超時(shí)時(shí)間和次數(shù)

盡快明確的向服務(wù)端返回成功或失敗，不要以超時(shí)（有時(shí)是異常拋出）代替消費(fèi)失敗。

不要用重試機(jī)制來進(jìn)行業(yè)務(wù)限流

錯(cuò)誤示例：如果當(dāng)前消費(fèi)速度過高觸發(fā)限流，則返回消費(fèi)失敗，等待下次重新消費(fèi)。正確示例：如果當(dāng)前消費(fèi)速度過高觸發(fā)限流，則延遲獲取消息，稍后再消費(fèi)。

發(fā)送重試和消費(fèi)重試會(huì)導(dǎo)致相同的消息重復(fù)消費(fèi)，消費(fèi)方應(yīng)該有一個(gè)良好的冪等設(shè)計(jì)

正確示例：某系統(tǒng)中消費(fèi)的邏輯是為某個(gè)用戶發(fā)送短信，該短信已經(jīng)發(fā)送成功了，當(dāng)消費(fèi)者應(yīng)用重復(fù)收到該消息，此時(shí)應(yīng)該返回消費(fèi)成功。

總結(jié)

Cloud Native

本文主要介紹重試的基本概念，生產(chǎn)者消費(fèi)者收發(fā)消息時(shí)觸發(fā)重試的條件和具體行為，以及 RocketMQ 收發(fā)容錯(cuò)的最佳實(shí)踐。重試策略幫助我們從隨機(jī)的、短暫的瞬態(tài)故障中恢復(fù)，是在容忍錯(cuò)誤時(shí)，提高可用性的一種強(qiáng)大機(jī)制。但請(qǐng)謹(jǐn)記 “重試是對(duì)于分布式系統(tǒng)來說自私的”，因?yàn)榭蛻舳苏J(rèn)為其請(qǐng)求很重要，并要求服務(wù)端花費(fèi)更多資源來處理，盲目的重試設(shè)計(jì)不可取，合理的使用重試可以幫助我們構(gòu)建更加彈性且可靠的系統(tǒng)。