RabbitMQ高級之失敗重試機制(含源碼)
作者：程序猿羊
當消費者出現(xiàn)異常后，消息會不斷requeue(重新入隊）到隊列，再重新發(fā)送給消費者，然后再次異常，再次requeue，無限循環(huán)，導致mq的消息處理飆升，帶來不必要的壓力。
一、失敗重試機制
當消費者出現(xiàn)異常后，消息會不斷requeue(重新入隊）到隊列，再重新發(fā)送給消費者，然后再次異常，再次requeue，無限循環(huán)，導致mq的消息處理飆升，帶來不必要的壓力。
為此，可利用Spring的retry機制，在消費者出現(xiàn)異常時利用本地重試，而不是無限制的requeue到mq隊列。
配置參數(shù)如下：
listener: # 開啟消費者確認其機制
simple:
prefetch: 1 #消費者每次只能獲取一條消息，處理完才能獲取下一條（可實現(xiàn)能者多勞）
acknowledge-mode: AUTO # none：關(guān)閉ack；manual：手動ack；auto：自動ack
retry:
enabled: true #開啟消費者失敗重試
initial-interval: 1000ms #初始的失敗等待時長為1秒
multiplier: 1 #下次失敗的等待時長倍數(shù)，下次等待時長 = multiplier * last-interval
max-attempts: 3 #最大重試次數(shù)
stateless: true #true無狀態(tài);false有狀態(tài)。如果業(yè)務中包含事務，這里改為false測試結(jié)果：
圖片
但是重試三次后，隊列里面的消息被踢出了：
圖片
二、失敗消息處理策略
在開啟重試模式后，重試次數(shù)耗盡，如果消息依然失敗，則需要有MessageRecoverer接口來處理，它包含三種不同的實現(xiàn):
1、RejectAndDontRequeueRecoverer:重試耗盡后，直接reject，丟棄消息。默認就是這種方式。
2、ImmediateRequeueMessageRecoverer:重試耗盡后，返回nack，消息重新入隊(不建議采用：會出現(xiàn)死循環(huán))。
3、RepublishMessageRecoverer:重試耗盡后，將失敗消息投遞到指定的交換機。(推薦使用)
圖片
三、實現(xiàn)方式
3.1、定義接收失敗消息的交換機、隊列及其綁定關(guān)系:
/**
* 功能描述:定義接收錯誤消費的日志
* @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ā)送錯誤的消息: " + message);
}3.2、定義RepublishMessageRecoverer:
/**
* @Description: TODO:定義錯誤消息接收
* @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、測試結(jié)果：
圖片
3.4、總結(jié)
消費者如何保證消息一定被消費?
- 開啟消費者確認機制為auto，由spring確認消息處理成功后返回ack，異常時返回nack。如果一直處理異常會一直重試。
- 開啟消費者失敗重試機制，并設(shè)置MessageRecoverer，多次重試失敗后將消息投遞到異常交換機，交由人工處理。
責任編輯：武曉燕
來源：
程序猿小楊
