RabbitMQ高級(jí)之失敗重試機(jī)制(含源碼)
作者：程序猿羊
當(dāng)消費(fèi)者出現(xiàn)異常后，消息會(huì)不斷requeue(重新入隊(duì)）到隊(duì)列，再重新發(fā)送給消費(fèi)者，然后再次異常，再次requeue，無限循環(huán)，導(dǎo)致mq的消息處理飆升，帶來不必要的壓力。
一、失敗重試機(jī)制
當(dāng)消費(fèi)者出現(xiàn)異常后，消息會(huì)不斷requeue(重新入隊(duì)）到隊(duì)列，再重新發(fā)送給消費(fèi)者，然后再次異常，再次requeue，無限循環(huán)，導(dǎo)致mq的消息處理飆升，帶來不必要的壓力。
為此，可利用Spring的retry機(jī)制，在消費(fèi)者出現(xiàn)異常時(shí)利用本地重試，而不是無限制的requeue到mq隊(duì)列。
配置參數(shù)如下：
listener: # 開啟消費(fèi)者確認(rèn)其機(jī)制
simple:
prefetch: 1 #消費(fèi)者每次只能獲取一條消息，處理完才能獲取下一條（可實(shí)現(xiàn)能者多勞）
acknowledge-mode: AUTO # none：關(guān)閉ack；manual：手動(dòng)ack；auto：自動(dòng)ack
retry:
enabled: true #開啟消費(fèi)者失敗重試
initial-interval: 1000ms #初始的失敗等待時(shí)長(zhǎng)為1秒
multiplier: 1 #下次失敗的等待時(shí)長(zhǎng)倍數(shù)，下次等待時(shí)長(zhǎng) = multiplier * last-interval
max-attempts: 3 #最大重試次數(shù)
stateless: true #true無狀態(tài);false有狀態(tài)。如果業(yè)務(wù)中包含事務(wù)，這里改為false測(cè)試結(jié)果：
圖片
但是重試三次后，隊(duì)列里面的消息被踢出了：
圖片
二、失敗消息處理策略
在開啟重試模式后，重試次數(shù)耗盡，如果消息依然失敗，則需要有MessageRecoverer接口來處理，它包含三種不同的實(shí)現(xiàn):
1、RejectAndDontRequeueRecoverer:重試耗盡后，直接reject，丟棄消息。默認(rèn)就是這種方式。
2、ImmediateRequeueMessageRecoverer:重試耗盡后，返回nack，消息重新入隊(duì)(不建議采用：會(huì)出現(xiàn)死循環(huán))。
3、RepublishMessageRecoverer:重試耗盡后，將失敗消息投遞到指定的交換機(jī)。(推薦使用)
圖片
三、實(shí)現(xiàn)方式
3.1、定義接收失敗消息的交換機(jī)、隊(duì)列及其綁定關(guān)系:
/**
* 功能描述:定義接收錯(cuò)誤消費(fèi)的日志
* @MethodName: receiveErrorMessage
* @MethodParam: [message]
* @Return: void
* @Author: yyalin
* @CreateDate: 2023/11/15 9:55
*/
@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "errorQueue"),
exchange = @Exchange(name = "errorExchange", type = ExchangeTypes.DIRECT, ignoreDeclarationExceptions = "true"),
key = "errorRouting"
))
public void receiveErrorMessage(String message) {
log.info("消費(fèi)者收到發(fā)送錯(cuò)誤的消息: " + message);
}3.2、定義RepublishMessageRecoverer:
/**
* @Description: TODO:定義錯(cuò)誤消息接收
* @Author: yyalin
* @CreateDate: 2023/11/15 9:58
* @Version: V1.0
*/
@Configuration
@Slf4j
public class ErrorConfig {
@Bean
public MessageRecoverer messageRecoverer(RabbitTemplate rabbitTemplate){
log.debug("加載RepublishMessageRecoverer");
return new RepublishMessageRecoverer(rabbitTemplate,"errorExchange","errorRouting");
}
}3.3、測(cè)試結(jié)果：
圖片
3.4、總結(jié)
消費(fèi)者如何保證消息一定被消費(fèi)?
- 開啟消費(fèi)者確認(rèn)機(jī)制為auto，由spring確認(rèn)消息處理成功后返回ack，異常時(shí)返回nack。如果一直處理異常會(huì)一直重試。
- 開啟消費(fèi)者失敗重試機(jī)制，并設(shè)置MessageRecoverer，多次重試失敗后將消息投遞到異常交換機(jī)，交由人工處理。
責(zé)任編輯：武曉燕
來源：
程序猿小楊
