[[384109]]

本文轉(zhuǎn)載自微信公眾號(hào)「Java極客技術(shù)」，作者鴨血粉絲。轉(zhuǎn)載本文請(qǐng)聯(lián)系Java極客技術(shù)公眾號(hào)。   

一、介紹

在介紹消息中間件 MQ 之前，我們先來(lái)簡(jiǎn)單的了解一下，為何要引用消息中間件。

例如，在電商平臺(tái)中，常見(jiàn)的用戶下單，會(huì)經(jīng)歷以下幾個(gè)流程。

當(dāng)用戶下單時(shí)，創(chuàng)建完訂單之后，會(huì)調(diào)用第三方支付平臺(tái)，對(duì)用戶的賬戶金額進(jìn)行扣款，如果平臺(tái)支付扣款成功，會(huì)將結(jié)果通知到對(duì)應(yīng)的業(yè)務(wù)系統(tǒng)，接著業(yè)務(wù)系統(tǒng)會(huì)更新訂單狀態(tài)，同時(shí)調(diào)用倉(cāng)庫(kù)接口，進(jìn)行減庫(kù)存，通知物流進(jìn)行發(fā)貨!

試想一下，從訂單狀態(tài)更新、到扣減庫(kù)存、通知物流發(fā)貨都在一個(gè)方法內(nèi)同步完成，假如用戶支付成功、訂單狀態(tài)更新也成功，但是在扣減庫(kù)存或者通知物流發(fā)貨步驟失敗了，那么就會(huì)造成一個(gè)問(wèn)題，用戶已經(jīng)支付成功了，只是在倉(cāng)庫(kù)扣減庫(kù)存方面失敗，從而導(dǎo)致整個(gè)交易失敗!

一單失敗，老板可以假裝看不見(jiàn)，但是如果上千個(gè)單子都因此失敗，那么因系統(tǒng)造成的業(yè)務(wù)損失，將是巨大的，老板可能坐不住了!

因此，針對(duì)這種業(yè)務(wù)場(chǎng)景，架構(gòu)師們引入了異步通信技術(shù)方案，從而保證服務(wù)的高可用，大體流程如下：

當(dāng)訂單系統(tǒng)收到支付平臺(tái)發(fā)送的扣款結(jié)果之后，會(huì)將訂單消息發(fā)送到 MQ 消息中間件，同時(shí)也會(huì)更新訂單狀態(tài)。

在另一端，由倉(cāng)庫(kù)系統(tǒng)來(lái)異步監(jiān)聽(tīng)訂單系統(tǒng)發(fā)送的消息，當(dāng)收到訂單消息之后，再操作扣減庫(kù)存、通知物流公司發(fā)貨等服務(wù)!

在優(yōu)化后的流程下，即使扣減庫(kù)存服務(wù)失敗，也不會(huì)影響用戶交易。

正如《人月神話》中所說(shuō)的，軟件工程，沒(méi)有銀彈!

當(dāng)引入了 MQ 消息中間件之后，同樣也會(huì)帶來(lái)另一個(gè)問(wèn)題，假如 MQ 消息中間件突然宕機(jī)了，導(dǎo)致消息無(wú)法發(fā)送出去，那倉(cāng)庫(kù)系統(tǒng)就無(wú)法接受到訂單消息，進(jìn)而也無(wú)法發(fā)貨!

針對(duì)這個(gè)問(wèn)題，業(yè)界主流的解決辦法是采用集群部署，一主多從模式，從而實(shí)現(xiàn)服務(wù)的高可用，即使一臺(tái)機(jī)器突然宕機(jī)了，也依然能保證服務(wù)可用，在服務(wù)器故障期間，通過(guò)運(yùn)維手段，將服務(wù)重新啟動(dòng)，之后服務(wù)依然能正常運(yùn)行!

但是還有另一個(gè)問(wèn)題，假如倉(cāng)庫(kù)系統(tǒng)已經(jīng)收到訂單消息了，但是業(yè)務(wù)處理異常，或者服務(wù)器異常，導(dǎo)致當(dāng)前商品庫(kù)存并沒(méi)有扣減，也沒(méi)有發(fā)貨!

這個(gè)時(shí)候又改如何處理呢?

今天我們所要介紹的正是這種場(chǎng)景，假如消息消費(fèi)失敗，我們應(yīng)該如何處理?

二、解決方案

針對(duì)消息消費(fèi)失敗的場(chǎng)景，我們一般會(huì)通過(guò)如下方式進(jìn)行處理：

• 當(dāng)消息消費(fèi)失敗時(shí)，會(huì)對(duì)消息進(jìn)行重新推送
• 如果重試次數(shù)超過(guò)最大值，會(huì)將異常消息存儲(chǔ)到數(shù)據(jù)庫(kù)，然后人工介入排查問(wèn)題，進(jìn)行手工重試

當(dāng)消息在客戶端消費(fèi)失敗時(shí)，我們會(huì)將異常的消息加入到一個(gè)消息重試對(duì)象中，同時(shí)設(shè)置最大重試次數(shù)，并將消息重新推送到 MQ 消息中間件里，當(dāng)重試次數(shù)超過(guò)最大值時(shí)，會(huì)將異常的消息存儲(chǔ)到 MongoDB數(shù)據(jù)庫(kù)中，方便后續(xù)查詢異常的信息。

基于以上系統(tǒng)模型，我們可以編寫一個(gè)公共重試組件，話不多說(shuō)，直接干!

三、代碼實(shí)踐

本次補(bǔ)償服務(wù)采用 rabbitmq 消息中間件進(jìn)行處理，其他消息中間件處理思路也類似!

3.1、創(chuàng)建一個(gè)消息重試實(shí)體類

@Data 
@EqualsAndHashCode(callSuper = false) 
@Accessors(chain = true) 
public class MessageRetryDTO implements Serializable { 
 
    private static final long serialVersionUID = 1L; 
 
    /** 
     * 原始消息body 
     */ 
    private String bodyMsg; 
 
    /** 
     * 消息來(lái)源ID 
     */ 
    private String sourceId; 
 
    /** 
     * 消息來(lái)源描述 
     */ 
    private String sourceDesc; 
 
    /** 
     * 交換器 
     */ 
    private String exchangeName; 
 
    /** 
     * 路由鍵 
     */ 
    private String routingKey; 
 
    /** 
     * 隊(duì)列 
     */ 
    private String queueName; 
 
    /** 
     * 狀態(tài),1:初始化，2：成功，3：失敗 
     */ 
    private Integer status = 1; 
 
    /** 
     * 最大重試次數(shù) 
     */ 
    private Integer maxTryCount = 3; 
 
    /** 
     * 當(dāng)前重試次數(shù) 
     */ 
    private Integer currentRetryCount = 0; 
 
    /** 
     * 重試時(shí)間間隔（毫秒） 
     */ 
    private Long retryIntervalTime = 0L; 
 
    /** 
     * 任務(wù)失敗信息 
     */ 
    private String errorMsg; 
 
    /** 
     * 創(chuàng)建時(shí)間 
     */ 
    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 + 
                '}'; 
    } 
 
    /** 
     * 檢查重試次數(shù)是否超過(guò)最大值 
     * 
     * @return 
     */ 
    public boolean checkRetryCount() { 
        retryCountCalculate(); 
        //檢查重試次數(shù)是否超過(guò)最大值 
        if (this.currentRetryCount < this.maxTryCount) { 
            return true; 
        } 
        return false; 
    } 
 
    /** 
     * 重新計(jì)算重試次數(shù) 
     */ 
    private void retryCountCalculate() { 
        this.currentRetryCount = this.currentRetryCount + 1; 
    } 
 
} 

3.2、編寫服務(wù)重試抽象類

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("{} 收到消息: {}，業(yè)務(wù)數(shù)據(jù)：{}", 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("處理消息異常，錯(cuò)誤信息：", e); 
        } 
    } 
 
    /** 
     * 準(zhǔn)備執(zhí)行 
     * 
     * @param retryDto 
     */ 
    protected void prepareAction(MessageRetryDTO retryDto) { 
        try { 
            execute(retryDto); 
            doSuccessCallBack(retryDto); 
        } catch (Exception e) { 
            log.error("當(dāng)前任務(wù)執(zhí)行異常，業(yè)務(wù)數(shù)據(jù)：" + retryDto.toString(), e); 
            //執(zhí)行失敗，計(jì)算是否還需要繼續(xù)重試 
            if (retryDto.checkRetryCount()) { 
                if (log.isInfoEnabled()) { 
                    log.info("重試消息:{}", retryDto.toString()); 
                } 
                retrySend(retryDto); 
            } else { 
                if (log.isWarnEnabled()) { 
                    log.warn("當(dāng)前任務(wù)重試次數(shù)已經(jīng)到達(dá)最大次數(shù)，業(yè)務(wù)數(shù)據(jù)：" + retryDto.toString(), e); 
                } 
                doFailCallBack(retryDto.setErrorMsg(e.getMessage())); 
            } 
        } 
    } 
 
    /** 
     * 任務(wù)執(zhí)行成功，回調(diào)服務(wù)(根據(jù)需要進(jìn)行重寫) 
     * 
     * @param messageRetryDto 
     */ 
    private void doSuccessCallBack(MessageRetryDTO messageRetryDto) { 
        try { 
            successCallback(messageRetryDto); 
        } catch (Exception e) { 
            log.warn("執(zhí)行成功回調(diào)異常，隊(duì)列描述：{}，錯(cuò)誤原因：{}", messageRetryDto.getSourceDesc(), e.getMessage()); 
        } 
    } 
 
    /** 
     * 任務(wù)執(zhí)行失敗，回調(diào)服務(wù)(根據(jù)需要進(jìn)行重寫) 
     * 
     * @param messageRetryDto 
     */ 
    private void doFailCallBack(MessageRetryDTO messageRetryDto) { 
        try { 
            saveMessageRetryInfo(messageRetryDto.setErrorMsg(messageRetryDto.getErrorMsg())); 
            failCallback(messageRetryDto); 
        } catch (Exception e) { 
            log.warn("執(zhí)行失敗回調(diào)異常，隊(duì)列描述：{}，錯(cuò)誤原因：{}", messageRetryDto.getSourceDesc(), e.getMessage()); 
        } 
    } 
 
    /** 
     * 執(zhí)行任務(wù) 
     * 
     * @param messageRetryDto 
     */ 
    protected abstract void execute(MessageRetryDTO messageRetryDto); 
 
    /** 
     * 成功回調(diào) 
     * 
     * @param messageRetryDto 
     */ 
    protected abstract void successCallback(MessageRetryDTO messageRetryDto); 
 
    /** 
     * 失敗回調(diào) 
     * 
     * @param messageRetryDto 
     */ 
    protected abstract void failCallback(MessageRetryDTO messageRetryDto); 
 
    /** 
     * 構(gòu)建消息補(bǔ)償實(shí)體 
     * @param message 
     * @return 
     */ 
    private MessageRetryDTO buildMessageRetryInfo(Message message){ 
        //如果頭部包含補(bǔ)償消息實(shí)體，直接返回 
        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); 
            } 
        } 
        //自動(dòng)將業(yè)務(wù)消息加入補(bǔ)償實(shí)體 
        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; 
    } 
 
    /** 
     * 異常消息重新入庫(kù) 
     * @param retryDto 
     */ 
    private void retrySend(MessageRetryDTO retryDto){ 
        //將補(bǔ)償消息實(shí)體放入頭部，原始消息內(nèi)容保持不變 
        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); 
    } 
 
 
 
    /** 
     * 將異常消息存儲(chǔ)到mongodb中 
     * @param retryDto 
     */ 
    private void saveMessageRetryInfo(MessageRetryDTO retryDto){ 
        try { 
            mongoTemplate.save(retryDto, "message_retry_info"); 
        } catch (Exception e){ 
            log.error("將異常消息存儲(chǔ)到mongodb失敗，消息數(shù)據(jù)：" + retryDto.toString(), e); 
        } 
    } 
} 

3.3、編寫監(jiān)聽(tīng)服務(wù)類

在消費(fèi)端應(yīng)用的時(shí)候，也非常簡(jiǎn)單，例如，針對(duì)扣減庫(kù)存操作，我們可以通過(guò)如下方式進(jìn)行處理!

@Component 
public class OrderServiceListener extends CommonMessageRetryService { 
 
    private static final Logger log = LoggerFactory.getLogger(OrderServiceListener.class); 
 
    /** 
     * 監(jiān)聽(tīng)訂單系統(tǒng)下單成功消息 
     * @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) { 
        //調(diào)用扣減庫(kù)存服務(wù)，將業(yè)務(wù)異常拋出來(lái) 
    } 
 
    @Override 
    protected void successCallback(MessageRetryDTO messageRetryDto) { 
        //業(yè)務(wù)處理成功，回調(diào) 
    } 
 
    @Override 
    protected void failCallback(MessageRetryDTO messageRetryDto) { 
        //業(yè)務(wù)處理失敗，回調(diào) 
    } 
} 

當(dāng)消息消費(fèi)失敗，并超過(guò)最大次數(shù)時(shí)，會(huì)將消息存儲(chǔ)到 mongodb 中，然后像常規(guī)數(shù)據(jù)庫(kù)操作一樣，可以通過(guò) web 接口查詢異常消息，并針對(duì)具體場(chǎng)景進(jìn)行重試!

四、小結(jié)

可能有的同學(xué)會(huì)問(wèn)，為啥不將異常消息存在數(shù)據(jù)庫(kù)?

起初的確是存儲(chǔ)在 MYSQL 中，但是隨著業(yè)務(wù)的快速發(fā)展，訂單消息數(shù)據(jù)結(jié)構(gòu)越來(lái)越復(fù)雜，數(shù)據(jù)量也非常的大，甚至大到 MYSQL 中的 text 類型都無(wú)法存儲(chǔ)，同時(shí)這種數(shù)據(jù)結(jié)構(gòu)也不太適合在 MYSQL 中存儲(chǔ)，因此將其遷移到 mongodb!

本文主要圍繞消息消費(fèi)失敗這種場(chǎng)景，進(jìn)行基礎(chǔ)的方案和代碼實(shí)踐講解，可能有理解不到位的地方，歡迎批評(píng)指出!

五、參考

1、石杉的架構(gòu)筆記 - 如何處理消息消費(fèi)失敗問(wèn)題