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

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

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

1、消息發(fā)送

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

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

（2）消息通過交換機exchange被路由到隊列queue;

消息只有被正確投遞到隊列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);//設置消息持久化
        CorrelationDataExt correlationData = new CorrelationDataExt();
        correlationData.setId(msgId);
        correlationData.setData(json);
        rabbitTemplate.setEncoding("UTF-8");
        rabbitTemplate.setMandatory(true);//設置手工ack確認
        rabbitTemplate.setConfirmCallback(this);//ack回調(diào)
        rabbitTemplate.setReturnCallback(this);//回退回調(diào)
        rabbitTemplate.convertAndSend(queueName, message, correlationData);
        return true;
    }

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

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

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

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

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

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

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

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

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

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

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

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

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

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

    @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個參數(shù)，correlationData是消息發(fā)送時攜帶的數(shù)據(jù)對象，ack消息是否成功發(fā)送到exchange，cause是發(fā)送失敗時的原因。

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

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

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

自定義CorrelationData對象：

/**
 * CorrelationData的自定義實現(xiàn)，用于拿到消息內(nèi)容
 */
public class CorrelationDataExt extends CorrelationData {
    //數(shù)據(jù)
    private volatile Object data;
    //隊列
    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ù)據(jù)對象
        CorrelationDataExt correlationData = new CorrelationDataExt();
        correlationData.setId(msgId);
        correlationData.setData(json);
        correlationData.setQueueName(queueName);

        rabbitTemplate.setEncoding("UTF-8");
        rabbitTemplate.setMandatory(true);//設置手工ack確認
        rabbitTemplate.setConfirmCallback(this);//設置發(fā)送成功回調(diào)
        rabbitTemplate.setReturnCallback(this);//設置消息回退回調(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進行處理：

    @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失敗

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

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

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

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