rabbitmq消息ACK確認(rèn)機(jī)制及發(fā)送失敗處理方式

更新時(shí)間：2023年12月15日 09:29:51 作者：古甲哈醒

這篇文章主要介紹了rabbitmq消息ACK確認(rèn)機(jī)制及發(fā)送失敗處理方式,具有很好的參考價(jià)值,希望對大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教

1、消息發(fā)送

rabbitmq消息發(fā)送分為兩個(gè)階段：

（1）producter將消息發(fā)送到broker，即發(fā)送到exchage交換機(jī)；

（2）消息通過交換機(jī)exchange被路由到隊(duì)列queue;

消息只有被正確投遞到隊(duì)列queue中，才算發(fā)送成功。

消息發(fā)送代碼：

    public boolean send(String queueName, String json, String msgId){
        Message message = MessageBuilder.withBody(json.getBytes()).setCorrelationId(msgId).build();
        message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);//設(shè)置消息持久化
        CorrelationDataExt correlationData = new CorrelationDataExt();
        correlationData.setId(msgId);
        correlationData.setData(json);
        rabbitTemplate.setEncoding("UTF-8");
        rabbitTemplate.setMandatory(true);//設(shè)置手工ack確認(rèn)
        rabbitTemplate.setConfirmCallback(this);//ack回調(diào)
        rabbitTemplate.setReturnCallback(this);//回退回調(diào)
        rabbitTemplate.convertAndSend(queueName, message, correlationData);
        return true;
    }

在消息發(fā)送之前，我們要設(shè)置ack機(jī)制相關(guān)參數(shù)：

setMandatory：設(shè)置手工確認(rèn)ack；
setConfirmCallback：設(shè)置消息發(fā)送到exchange結(jié)果回調(diào)；
setReturnCallback：設(shè)置消息投遞到queue失敗回退時(shí)回調(diào)；

通過上述兩個(gè)回調(diào)方法，我們能夠?qū)Πl(fā)送失敗的消息進(jìn)行重發(fā)處理，確保消息不丟失。

2、消息發(fā)送失敗

根據(jù)rabbitmq發(fā)送過程，消息發(fā)送失敗的有三種情況會(huì)出現(xiàn)：

（1）producter連接mq失敗，消息沒有發(fā)送到mq

（2）producter連接mq成功，但是發(fā)送到exchange失敗

（3）消息發(fā)送到exchange成功，但是路由到queue失??；

3、發(fā)送失敗處理

（1）producter連接mq失敗，消息沒有發(fā)送到mq

這種情況下，在發(fā)送消息時(shí)可以通過捕捉AmqpException異常，將消息保存db中后續(xù)進(jìn)行重發(fā)處理。

        try{
            rabbitTemplate.convertAndSend(queueName, message, correlationData);
        }catch (Exception e){
            logger.error("連接MQ失敗", e);
            //todo 存儲(chǔ)到db中進(jìn)行重發(fā)
        }

（2）producter連接mq成功，但是發(fā)送到exchange失敗

通過實(shí)現(xiàn)ConfirmCallback接口，對發(fā)送結(jié)果進(jìn)行處理。

    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        String msgId = correlationData.getId();
        if(ack){
            //發(fā)送成功
            logger.debug("ack,消息投遞到exchange成功,msgId:{}",msgId);
        }else{
            //發(fā)送失敗，重試
            logger.error("ack,消息投遞exchange失敗，msgId:{},原因{}" ,msgId, cause);
           
        }
    }

confirm方法有3個(gè)參數(shù)，correlationData是消息發(fā)送時(shí)攜帶的數(shù)據(jù)對象，ack消息是否成功發(fā)送到exchange，cause是發(fā)送失敗時(shí)的原因。

通過ack我們可以判斷發(fā)送到exchange是否成功，如果ack=false，則我們進(jìn)行失敗處理。

但是這里存在一個(gè)問題，correlationData里面只有一個(gè)id屬性，沒有關(guān)于消息內(nèi)容的屬性，對于數(shù)據(jù)失敗處理非常不方便。

為解決此問題，我們可以自定義一個(gè)CorrelationData擴(kuò)展對象，繼承CorrelationData，并添加自己想要保存數(shù)據(jù)的屬性，在消息發(fā)送時(shí)，攜帶相關(guān)數(shù)據(jù)在該對象上即可。

自定義CorrelationData對象：

/**
 * CorrelationData的自定義實(shí)現(xiàn)，用于拿到消息內(nèi)容
 */
public class CorrelationDataExt extends CorrelationData {
    //數(shù)據(jù)
    private volatile Object data;
    //隊(duì)列
    private String queueName;

    public Object getData() {
        return data;
    }

    public void setData(Object data) {
        this.data = data;
    }

    public String getQueueName() {
        return queueName;
    }

    public void setQueueName(String queueName) {
        this.queueName = queueName;
    }
}

重寫發(fā)送方法，使用CorrelationDataExt對象攜帶數(shù)據(jù)：

    public boolean send(String queueName, String json, String msgId){
        Message message = MessageBuilder.withBody(json.getBytes()).setCorrelationId(msgId).build();
        message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);//設(shè)置消息持久化

        //使用自定義的數(shù)據(jù)對象
        CorrelationDataExt correlationData = new CorrelationDataExt();
        correlationData.setId(msgId);
        correlationData.setData(json);
        correlationData.setQueueName(queueName);

        rabbitTemplate.setEncoding("UTF-8");
        rabbitTemplate.setMandatory(true);//設(shè)置手工ack確認(rèn)
        rabbitTemplate.setConfirmCallback(this);//設(shè)置發(fā)送成功回調(diào)
        rabbitTemplate.setReturnCallback(this);//設(shè)置消息回退回調(diào)
        try{
            rabbitTemplate.convertAndSend(queueName, message, correlationData);//使用amqp default exchange direct
        }catch (Exception e){
            logger.error("MQ連接失敗，請聯(lián)系管理員處理!!!!");
            //保存到db重發(fā)
            saveToDB(msgId, json, queueName, "90");
        }
        return true;
    }

重寫confirm方法，對CorrelationData進(jìn)行處理：

    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        String msgId = correlationData.getId();
        if(ack){
            //發(fā)送成功
            logger.debug("ack,消息投遞到exchange成功,msgId:{}",msgId);
        }else{
            //發(fā)送失敗，重試
            logger.error("ack,消息投遞exchange失敗，msgId:{},原因{}" ,msgId, cause);
            if(correlationData instanceof CorrelationDataExt){
                CorrelationDataExt correlationDataExt = (CorrelationDataExt) correlationData;
                String message = (String) correlationDataExt.getData();
                String queueName = ((CorrelationDataExt) correlationData).getQueueName();
                saveToDB(msgId, message, queueName, "91");
            }else{
                logger.info("correlationData對象不包含數(shù)據(jù)");
            }
        }
    }

（3）消息發(fā)送到exchange成功，但是路由到queue失敗

通過實(shí)現(xiàn)ReturnCallback接口，對回退消息進(jìn)行重發(fā)處理。

    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        logger.error("消息發(fā)送失敗-消息回退，應(yīng)答碼：{}，原因：{}，交換機(jī)：{}，路由鍵：{}", replyCode, replyText, exchange, routingKey);
        String msgId = message.getMessageProperties().getCorrelationId();
        String data = new String(message.getBody());
        saveToDB(msgId, data, routingKey, "92");
    }

關(guān)于對失敗消息的處理，我這里是統(tǒng)一保存到DB中，后續(xù)通過定時(shí)任務(wù)進(jìn)行重發(fā)處理的。

通過以上3個(gè)方面對失敗消息的處理，可以確保消息能夠成功發(fā)送到mq，確保不丟失。