如何處理消息消費失敗

本篇內容介紹了“如何處理消息消費失敗”的有關知識，在實際案例的操作過程中，不少人都會遇到這樣的困境，接下來就讓小編帶領大家學習一下如何處理這些情況吧！希望大家仔細閱讀，能夠學有所成！

一、介紹

在介紹消息中間件 MQ 之前，我們先來簡單的了解一下，為何要引用消息中間件。

例如，在電商平臺中，常見的用戶下單，會經歷以下幾個流程。

當用戶下單時，創建完訂單之后，會調用第三方支付平臺，對用戶的賬戶金額進行扣款，如果平臺支付扣款成功，會將結果通知到對應的業務系統，接著業務系統會更新訂單狀態，同時調用倉庫接口，進行減庫存，通知物流進行發貨!

試想一下，從訂單狀態更新、到扣減庫存、通知物流發貨都在一個方法內同步完成，假如用戶支付成功、訂單狀態更新也成功，但是在扣減庫存或者通知物流發貨步驟失敗了，那么就會造成一個問題，用戶已經支付成功了，只是在倉庫扣減庫存方面失敗，從而導致整個交易失敗!

一單失敗，老板可以假裝看不見，但是如果上千個單子都因此失敗，那么因系統造成的業務損失，將是巨大的，老板可能坐不住了!

因此，針對這種業務場景，架構師們引入了異步通信技術方案，從而保證服務的高可用，大體流程如下：

當訂單系統收到支付平臺發送的扣款結果之后，會將訂單消息發送到 MQ 消息中間件，同時也會更新訂單狀態。

在另一端，由倉庫系統來異步監聽訂單系統發送的消息，當收到訂單消息之后，再操作扣減庫存、通知物流公司發貨等服務!

在優化后的流程下，即使扣減庫存服務失敗，也不會影響用戶交易。

正如《人月神話》中所說的，軟件工程，沒有銀彈!

當引入了 MQ 消息中間件之后，同樣也會帶來另一個問題，假如 MQ  消息中間件突然宕機了，導致消息無法發送出去，那倉庫系統就無法接受到訂單消息，進而也無法發貨!

針對這個問題，業界主流的解決辦法是采用集群部署，一主多從模式，從而實現服務的高可用，即使一臺機器突然宕機了，也依然能保證服務可用，在服務器故障期間，通過運維手段，將服務重新啟動，之后服務依然能正常運行!

但是還有另一個問題，假如倉庫系統已經收到訂單消息了，但是業務處理異常，或者服務器異常，導致當前商品庫存并沒有扣減，也沒有發貨!

這個時候又改如何處理呢?

今天我們所要介紹的正是這種場景，假如消息消費失敗，我們應該如何處理?

二、解決方案

針對消息消費失敗的場景，我們一般會通過如下方式進行處理：

• 當消息消費失敗時，會對消息進行重新推送

• 如果重試次數超過最大值，會將異常消息存儲到數據庫，然后人工介入排查問題，進行手工重試

當消息在客戶端消費失敗時，我們會將異常的消息加入到一個消息重試對象中，同時設置最大重試次數，并將消息重新推送到 MQ  消息中間件里，當重試次數超過最大值時，會將異常的消息存儲到 MongoDB數據庫中，方便后續查詢異常的信息。

基于以上系統模型，我們可以編寫一個公共重試組件，話不多說，直接干!

三、代碼實踐

本次補償服務采用 rabbitmq 消息中間件進行處理，其他消息中間件處理思路也類似!

3.1、創建一個消息重試實體類

@Data @EqualsAndHashCode(callSuper = false) @Accessors(chain = true) public class MessageRetryDTO implements Serializable {      private static final long serialVersionUID = 1L;      /**      * 原始消息body      */     private String bodyMsg;      /**      * 消息來源ID      */     private String sourceId;      /**      * 消息來源描述      */     private String sourceDesc;      /**      * 交換器      */     private String exchangeName;      /**      * 路由鍵      */     private String routingKey;      /**      * 隊列      */     private String queueName;      /**      * 狀態,1:初始化，2：成功，3：失敗      */     private Integer status = 1;      /**      * 最大重試次數      */     private Integer maxTryCount = 3;      /**      * 當前重試次數      */     private Integer currentRetryCount = 0;      /**      * 重試時間間隔（毫秒）      */     private Long retryIntervalTime = 0L;      /**      * 任務失敗信息      */     private String errorMsg;      /**      * 創建時間      */     private Date createTime;      @Override     public String toString() {         return "MessageRetryDTO{" +                 "bodyMsg='" + bodyMsg + '\'' +                 ", sourceId='" + sourceId + '\'' +                 ", sourceDesc='" + sourceDesc + '\'' +                 ", exchangeName='" + exchangeName + '\'' +                 ", routingKey='" + routingKey + '\'' +                 ", queueName='" + queueName + '\'' +                 ", status=" + status +                 ", maxTryCount=" + maxTryCount +                 ", currentRetryCount=" + currentRetryCount +                 ", retryIntervalTime=" + retryIntervalTime +                 ", errorMsg='" + errorMsg + '\'' +                 ", createTime=" + createTime +                 '}';     }      /**      * 檢查重試次數是否超過最大值      *      * @return      */     public boolean checkRetryCount() {         retryCountCalculate();         //檢查重試次數是否超過最大值         if (this.currentRetryCount < this.maxTryCount) {             return true;         }         return false;     }      /**      * 重新計算重試次數      */     private void retryCountCalculate() {         this.currentRetryCount = this.currentRetryCount + 1;     }  }

3.2、編寫服務重試抽象類

public abstract class CommonMessageRetryService {      private static final Logger log = LoggerFactory.getLogger(CommonMessageRetryService.class);      @Autowired     private RabbitTemplate rabbitTemplate;      @Autowired     private MongoTemplate mongoTemplate;       /**      * 初始化消息      *      * @param message      */     public void initMessage(Message message) {         log.info("{} 收到消息: {}，業務數據：{}", this.getClass().getName(), message.toString(), new String(message.getBody()));         try {             //封裝消息             MessageRetryDTO messageRetryDto = buildMessageRetryInfo(message);             if (log.isInfoEnabled()) {                 log.info("反序列化消息:{}", messageRetryDto.toString());             }             prepareAction(messageRetryDto);         } catch (Exception e) {             log.warn("處理消息異常，錯誤信息：", e);         }     }      /**      * 準備執行      *      * @param retryDto      */     protected void prepareAction(MessageRetryDTO retryDto) {         try {             execute(retryDto);             doSuccessCallBack(retryDto);         } catch (Exception e) {             log.error("當前任務執行異常，業務數據：" + retryDto.toString(), e);             //執行失敗，計算是否還需要繼續重試             if (retryDto.checkRetryCount()) {                 if (log.isInfoEnabled()) {                     log.info("重試消息:{}", retryDto.toString());                 }                 retrySend(retryDto);             } else {                 if (log.isWarnEnabled()) {                     log.warn("當前任務重試次數已經到達最大次數，業務數據：" + retryDto.toString(), e);                 }                 doFailCallBack(retryDto.setErrorMsg(e.getMessage()));             }         }     }      /**      * 任務執行成功，回調服務(根據需要進行重寫)      *      * @param messageRetryDto      */     private void doSuccessCallBack(MessageRetryDTO messageRetryDto) {         try {             successCallback(messageRetryDto);         } catch (Exception e) {             log.warn("執行成功回調異常，隊列描述：{}，錯誤原因：{}", messageRetryDto.getSourceDesc(), e.getMessage());         }     }      /**      * 任務執行失敗，回調服務(根據需要進行重寫)      *      * @param messageRetryDto      */     private void doFailCallBack(MessageRetryDTO messageRetryDto) {         try {             saveMessageRetryInfo(messageRetryDto.setErrorMsg(messageRetryDto.getErrorMsg()));             failCallback(messageRetryDto);         } catch (Exception e) {             log.warn("執行失敗回調異常，隊列描述：{}，錯誤原因：{}", messageRetryDto.getSourceDesc(), e.getMessage());         }     }      /**      * 執行任務      *      * @param messageRetryDto      */     protected abstract void execute(MessageRetryDTO messageRetryDto);      /**      * 成功回調      *      * @param messageRetryDto      */     protected abstract void successCallback(MessageRetryDTO messageRetryDto);      /**      * 失敗回調      *      * @param messageRetryDto      */     protected abstract void failCallback(MessageRetryDTO messageRetryDto);      /**      * 構建消息補償實體      * @param message      * @return      */     private MessageRetryDTO buildMessageRetryInfo(Message message){         //如果頭部包含補償消息實體，直接返回         Map<String, Object> messageHeaders = message.getMessageProperties().getHeaders();         if(messageHeaders.containsKey("message_retry_info")){             Object retryMsg = messageHeaders.get("message_retry_info");             if(Objects.nonNull(retryMsg)){                 return JSONObject.parseObject(String.valueOf(retryMsg), MessageRetryDTO.class);             }         }         //自動將業務消息加入補償實體         MessageRetryDTO messageRetryDto = new MessageRetryDTO();         messageRetryDto.setBodyMsg(new String(message.getBody(), StandardCharsets.UTF_8));         messageRetryDto.setExchangeName(message.getMessageProperties().getReceivedExchange());         messageRetryDto.setRoutingKey(message.getMessageProperties().getReceivedRoutingKey());         messageRetryDto.setQueueName(message.getMessageProperties().getConsumerQueue());         messageRetryDto.setCreateTime(new Date());         return messageRetryDto;     }      /**      * 異常消息重新入庫      * @param retryDto      */     private void retrySend(MessageRetryDTO retryDto){         //將補償消息實體放入頭部，原始消息內容保持不變         MessageProperties messageProperties = new MessageProperties();         messageProperties.setContentType(MessageProperties.CONTENT_TYPE_JSON);         messageProperties.setHeader("message_retry_info", JSONObject.toJSON(retryDto));         Message message = new Message(retryDto.getBodyMsg().getBytes(), messageProperties);         rabbitTemplate.convertAndSend(retryDto.getExchangeName(), retryDto.getRoutingKey(), message);     }        /**      * 將異常消息存儲到mongodb中      * @param retryDto      */     private void saveMessageRetryInfo(MessageRetryDTO retryDto){         try {             mongoTemplate.save(retryDto, "message_retry_info");         } catch (Exception e){             log.error("將異常消息存儲到mongodb失敗，消息數據：" + retryDto.toString(), e);         }     } }

3.3、編寫監聽服務類

在消費端應用的時候，也非常簡單，例如，針對扣減庫存操作，我們可以通過如下方式進行處理!

@Component public class OrderServiceListener extends CommonMessageRetryService {      private static final Logger log = LoggerFactory.getLogger(OrderServiceListener.class);      /**      * 監聽訂單系統下單成功消息      * @param message      */     @RabbitListener(queues = "mq.order.add")     public void consume(Message message) {         log.info("收到訂單下單成功消息: {}", message.toString());         super.initMessage(message);     }       @Override     protected void execute(MessageRetryDTO messageRetryDto) {         //調用扣減庫存服務，將業務異常拋出來     }      @Override     protected void successCallback(MessageRetryDTO messageRetryDto) {         //業務處理成功，回調     }      @Override     protected void failCallback(MessageRetryDTO messageRetryDto) {         //業務處理失敗，回調     } }

當消息消費失敗，并超過最大次數時，會將消息存儲到 mongodb 中，然后像常規數據庫操作一樣，可以通過 web  接口查詢異常消息，并針對具體場景進行重試!

“如何處理消息消費失敗”的內容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業相關的知識可以關注億速云網站，小編將為大家輸出更多高質量的實用文章！