脚本之家 服务器常用软件
快捷导航
您的位置：首页 → 软件编程 → java → rabbitmq消息ACK确认机制及发送失败处理

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

 更新时间：2023年12月15日 09:29:51   作者：古甲哈醒  
这篇文章主要介绍了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机制相关参数： 

  • setMandatory：设置手工确认ack；
  • setConfirmCallback：设置消息发送到exchange结果回调；
  • setReturnCallback：设置消息投递到queue失败回退时回调；

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

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，确保不丢失。

总结

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

您可能感兴趣的文章:

相关文章

最新评论