java

关注公众号 jb51net

关闭
首页 > 软件编程 > java > rabbitmq消息ACK确认机制及发送失败处理

rabbitmq消息ACK确认机制及发送失败处理方式

作者:古甲哈醒

这篇文章主要介绍了rabbitmq消息ACK确认机制及发送失败处理方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教

rabbitmq为确保消息发送和接收成功,采用ack机制。

(1)生产者producter发送消息到mq时,mq会发送ack给producter告知消息是否投递成功;

(2)消费者consumer接收处理消息后,consumer会发送ack给mq告知消息是否处理成功;

通过ack机制,确保消息能够被producter成功发送和consumer成功接收处理,保证消息不丢失。

1、消息发送

rabbitmq消息发送分为两个阶段:

(1)producter将消息发送到broker,即发送到exchage交换机;

(2)消息通过交换机exchange被路由到队列queue;

消息只有被正确投递到队列queue中,才算发送成功。

消息发送代码:

    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回调
        rabbitTemplate.setReturnCallback(this);//回退回调
        rabbitTemplate.convertAndSend(queueName, message, correlationData);
        return true;
    }

在消息发送之前,我们要设置ack机制相关参数: 

通过上述两个回调方法,我们能够对发送失败的消息进行重发处理,确保消息不丢失。

2、消息发送失败

根据rabbitmq发送过程,消息发送失败的有三种情况会出现:

(1)producter连接mq失败,消息没有发送到mq

(2)producter连接mq成功,但是发送到exchange失败

(3)消息发送到exchange成功,但是路由到queue失败;

3、发送失败处理

(1)producter连接mq失败,消息没有发送到mq

这种情况下,在发送消息时可以通过捕捉AmqpException异常,将消息保存db中后续进行重发处理。

        try{
            rabbitTemplate.convertAndSend(queueName, message, correlationData);
        }catch (Exception e){
            logger.error("连接MQ失败", e);
            //todo 存储到db中进行重发
        }

(2)producter连接mq成功,但是发送到exchange失败

通过实现ConfirmCallback接口,对发送结果进行处理。

    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        String msgId = correlationData.getId();
        if(ack){
            //发送成功
            logger.debug("ack,消息投递到exchange成功,msgId:{}",msgId);
        }else{
            //发送失败,重试
            logger.error("ack,消息投递exchange失败,msgId:{},原因{}" ,msgId, cause);
           
        }
    }

confirm方法有3个参数,correlationData是消息发送时携带的数据对象,ack消息是否成功发送到exchange,cause是发送失败时的原因。

通过ack我们可以判断发送到exchange是否成功,如果ack=false,则我们进行失败处理。 

但是这里存在一个问题,correlationData里面只有一个id属性,没有关于消息内容的属性,对于数据失败处理非常不方便。

为解决此问题,我们可以自定义一个CorrelationData扩展对象,继承CorrelationData,并添加自己想要保存数据的属性,在消息发送时,携带相关数据在该对象上即可

自定义CorrelationData对象:

/**
 * CorrelationData的自定义实现,用于拿到消息内容
 */
public class CorrelationDataExt extends CorrelationData {
    //数据
    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;
    }
}

重写发送方法,使用CorrelationDataExt对象携带数据:

    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);
        correlationData.setQueueName(queueName);

        rabbitTemplate.setEncoding("UTF-8");
        rabbitTemplate.setMandatory(true);//设置手工ack确认
        rabbitTemplate.setConfirmCallback(this);//设置发送成功回调
        rabbitTemplate.setReturnCallback(this);//设置消息回退回调
        try{
            rabbitTemplate.convertAndSend(queueName, message, correlationData);//使用amqp default exchange direct
        }catch (Exception e){
            logger.error("MQ连接失败,请联系管理员处理!!!!");
            //保存到db重发
            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){
            //发送成功
            logger.debug("ack,消息投递到exchange成功,msgId:{}",msgId);
        }else{
            //发送失败,重试
            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对象不包含数据");
            }
        }
    }

(3)消息发送到exchange成功,但是路由到queue失败

通过实现ReturnCallback接口,对回退消息进行重发处理。

    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        logger.error("消息发送失败-消息回退,应答码:{},原因:{},交换机:{},路由键:{}", replyCode, replyText, exchange, routingKey);
        String msgId = message.getMessageProperties().getCorrelationId();
        String data = new String(message.getBody());
        saveToDB(msgId, data, routingKey, "92");
    }

关于对失败消息的处理,我这里是统一保存到DB中,后续通过定时任务进行重发处理的。

通过以上3个方面对失败消息的处理,可以确保消息能够成功发送到mq,确保不丢失。

总结

以上为个人经验,希望能给大家一个参考,也希望大家多多支持脚本之家。

您可能感兴趣的文章:
阅读全文