rabbitMq消息确认机制 - 《田云的博客笔记》

导图
代码示例

导图

语雀内容
初始化队列的时候设置持久化。
provider

配置confirm模式，并编写confirm函数
发送消息的时候，设置 deliveryMode为持久化

consumer

设置ack模式
接收消息时，手动ack

代码示例

参考：https://blog.csdn.net/Weixiaohuai/article/details/94961012

不同版本代码可能不太一样，差别不大，例如在传递id的时候需要封装的参数有修改。

下面通过一个示例说明在RabbitMQ中实现发送确认和消费者确认的使用方法。

【a】引入pom.xml依赖：

1. <dependency>
2. <groupId>org.springframework.boot</groupId>
3. <artifactId>spring-boot-starter-amqp</artifactId>
4. </dependency>

【b】配置文件application.yml：

server:
  port: 6666
spring:
  application:
    name: mq-message-confirm2
  rabbitmq:
    host: 127.0.0.1
    virtual-host: /vhost
    username: wsh
    password: wsh
    port: 5672
    #消息发送确认回调
    publisher-confirms: true
    #指定消息确认模式为手动确认
    listener:
      simple:
        acknowledge-mode: manual
    #发送返回监听回调
    publisher-returns: true

这里需要注意两点：

spring.rabbitmq.publisher-confirms = true
spring.rabbitmq.listener.simple.acknowledge-mode = manual

【c】自定义消息发送、返回回调监听类

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import java.nio.charset.StandardCharsets;
/**
 * @Description 自定义消息发送确认的回调
 * @Author weishihuai
 * @Date 2019/6/27 10:42
 * <p>
 * 实现接口：implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback
 * ConfirmCallback：只确认消息是否正确到达交换机中,不管是否到达交换机,该回调都会执行;
 * ReturnCallback：如果消息从交换机未正确到达队列中将会执行，正确到达则不执行;
 */
@Component
public class CustomConfirmAndReturnCallback implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {
    private static final Logger logger = LoggerFactory.getLogger(CustomConfirmAndReturnCallback.class);
    @Autowired
    private RabbitTemplate rabbitTemplate;
    /**
     * PostConstruct: 用于在依赖关系注入完成之后需要执行的方法上，以执行任何初始化.
     */
    @PostConstruct
    public void init() {
        //指定 ConfirmCallback
        rabbitTemplate.setConfirmCallback(this);
        //指定 ReturnCallback
        rabbitTemplate.setReturnCallback(this);
    }
    /**
     * 消息从交换机成功到达队列，则returnedMessage方法不会执行;
     * 消息从交换机未能成功到达队列，则returnedMessage方法会执行;
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        logger.info("returnedMessage回调方法>>>" + new String(message.getBody(), StandardCharsets.UTF_8) + ",replyCode:" + replyCode
                + ",replyText:" + replyText + ",exchange:" + exchange + ",routingKey:" + routingKey);
    }
    /**
     * 如果消息没有到达交换机,则该方法中isSendSuccess = false,error为错误信息;
     * 如果消息正确到达交换机,则该方法中isSendSuccess = true;
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean isSendSuccess, String error) {
        logger.info("confirm回调方法>>>回调消息ID为: " + correlationData.getId());
        if (isSendSuccess) {
            logger.info("confirm回调方法>>>消息发送到交换机成功！");
        } else {
            logger.info("confirm回调方法>>>消息发送到交换机失败！，原因 : [{}]", error);
        }
    }
}

注意这里我同时也实现了RabbitTemplate.ReturnCallback返回回调接口，并且重写了returnedMessage()方法，返回回调主要指的是：如果消息从交换机未正确到达队列中将会执行，正确到达则不执行returnedMessage()。

【d】RabbitMQ配置信息

/**
 * @Description: RabbitMQ配置信息，绑定交换器、队列、路由键设置
 * @author: weishihuai
 * @Date: 2019/6/27 10:38
 * <p>
 * 说明：
 * <p>
 * 1. 声明Exchange交换器；
 * 2. 声明Queue队列；
 * 3. 绑定：BindingBuilder绑定队列到交换器，并设置路由键；
 */
@Component
public class RabbitMQConfig {
    private static final String EXCHANGE_NAME = "message_confirm_exchange";
    private static final String QUEUE_NAME = "message_confirm_queue";
    private static final String ROUTING_KEY = "user.#";
    @Bean
    private TopicExchange topicExchange() {
        return new TopicExchange(EXCHANGE_NAME);
    }
    @Bean
    private Queue queue() {
        return new Queue(QUEUE_NAME);
    }
    @Bean
    private Binding bindingDirect() {
        return BindingBuilder.bind(queue()).to(topicExchange()).with(ROUTING_KEY);
    }
}

【e】发送者：这里发送了三条消息，如果三条消息中某条消息已经被拒绝过一次，那么触发basicNack()重新回到队列中，如果该消息再次被拒绝，那么消费者将会调用basicReject()直接拒绝该条消息,以后也不会再次接收该消息。

/**
 * @Description: 生产者
 * @author: weishihuai
 * @Date: 2019/6/27 10:39
 */
@Component
public class Producer {
    private static final Logger logger = LoggerFactory.getLogger(Producer.class);
    private static final String EXCHANGE_NAME = "message_confirm_exchange";
    private static final String ROUTING_KEY = "user.add.submit";
    @Autowired
    public RabbitTemplate rabbitTemplate;
    public void sendMessage() {
        for (int i = 1; i <= 3; i++) {
            CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
            logger.info("【Producer】发送的消费ID = {}", correlationData.getId());
            String msg = "hello confirm message" + i;
            logger.info("【Producer】发送的消息 = {}", msg);
            rabbitTemplate.convertAndSend(EXCHANGE_NAME, ROUTING_KEY, msg, correlationData);
        }
    }
}

【f】消费者1：这里模拟了在处理消息的时候触发一个空指针异常，用于触发拒绝某个消息。

import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.io.IOException;
/**
 * @Description: 消费者1
 * @author: weishihuai
 * @Date: 2019/6/27 10:42
 */
@Component
public class Consumer01 {
    private static final Logger logger = LoggerFactory.getLogger(Consumer01.class);
    @RabbitListener(queues = "message_confirm_queue")
    public void receiveMessage01(String msg, Channel channel, Message message) throws IOException {
        try {
            // 这里模拟一个空指针异常，
            String string = null;
            string.length();
            logger.info("【Consumer01成功接收到消息】>>> {}", msg);
            // 确认收到消息，只确认当前消费者的一个消息收到
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            if (message.getMessageProperties().getRedelivered()) {
                logger.info("【Consumer01】消息已经回滚过，拒绝接收消息 ： {}", msg);
                // 拒绝消息，并且不再重新进入队列
                //public void basicReject(long deliveryTag, boolean requeue)
                channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
            } else {
                logger.info("【Consumer01】消息即将返回队列重新处理 ：{}", msg);
                //设置消息重新回到队列处理
                // requeue表示是否重新回到队列，true重新入队
                //public void basicNack(long deliveryTag, boolean multiple, boolean requeue)
                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
            }
            e.printStackTrace();
        }
    }
}

【g】消费者2：该消费者为正常消费消息。

import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.io.IOException;
/**
 * @Description: 消费者2
 * @author: weishihuai
 * @Date: 2019/6/27 10:42
 */
@Component
public class Consumer02 {
    private static final Logger logger = LoggerFactory.getLogger(Consumer02.class);
    @RabbitListener(queues = "message_confirm_queue")
    public void receiveMessage02(String msg, Channel channel, Message message) throws IOException {
        try {
            logger.info("【Consumer02成功接收到消息】>>> {}", msg);
            // 确认收到消息，只确认当前消费者的一个消息收到
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            if (message.getMessageProperties().getRedelivered()) {
                logger.info("【Consumer02】消息已经回滚过，拒绝接收消息 ： {}", msg);
                // 拒绝消息，并且不再重新进入队列
                //public void basicReject(long deliveryTag, boolean requeue)
                channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
            } else {
                logger.info("【Consumer02】消息即将返回队列重新处理 ：{}", msg);
                //设置消息重新回到队列处理
                // requeue表示是否重新回到队列，true重新入队
                //public void basicNack(long deliveryTag, boolean multiple, boolean requeue)
                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
            }
            e.printStackTrace();
        }
    }
}

【h】启动项目，查看运行结果

rabbitMq消息确认机制 - 图1
rabbitMq消息确认机制 - 图2
rabbitMq消息确认机制 - 图3

由控制台结果可见，hello confirm message1这条消息第一次被consumer1拒绝了一次，执行basicNack重新回到队列，第二次又被判断为该条消息已经回滚过，调用basicReject方法又被拒绝并且禁止重新回到队列，这样该条消息将不会被消费者重新消费。
hello confirm message2这条消息成功被consumer2消费掉，hello confirm message3这条消息第一次也被 consumer1拒绝了，但是在重新回到队列之后，被consumer2成功消费了。
同时可以看到，三条消息都正确从发送者到达交换机，所以都执行了 confirm(CorrelationData correlationData, boolean isSendSuccess, String error)回调方法。
因为消息都成功从交换机正确到达队列中，所有监听的returnCallback的returnedMessage()方法并没有被执行。下面我们测试一下，假设指定一个binding key不匹配的。修改下面的路由键，让消息无法从交换机正确路由到队列上：

rabbitMq消息确认机制 - 图4

首先在RabbitMQ管理控制台将之前的user.#绑定键解除绑定：
rabbitMq消息确认机制 - 图5

重新启动项目,查看控制台日志：

rabbitMq消息确认机制 - 图6

可见，三条消息都未能正确从交换机路由到队列，所以都执行了returnedMessage回调方法。

下面我们测试一下消息从发送者未能正确到达交换机的情况，这里主要修改一个不存在的交换机名称，这样消息就不能正确到达消费者监听队列所在的交换机message_confirm_exchange，从而触发confirmCallback中发送失败的情况，error为错误原因。

rabbitMq消息确认机制 - 图7

控制台日志：

1. 2019-07-07 11:22:27.762  INFO 12528 --- [ 127.0.0.1:5672] c.w.s.s.c.CustomConfirmAndReturnCallback : confirm回调方法>>>回调消息ID为: 01ee6352-861e-4d96-8346-d808508ae3d2
2. 2019-07-07 11:22:27.762  INFO 12528 --- [ 127.0.0.1:5672] c.w.s.s.c.CustomConfirmAndReturnCallback : confirm回调方法>>>消息发送到交换机失败！，原因 : [channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'message_confirm_exchange2' in vhost '/vhost', class-id=60, method-id=40)]
3. 2019-07-07 11:22:27.762  INFO 12528 --- [ 127.0.0.1:5672] c.w.s.s.c.CustomConfirmAndReturnCallback : confirm回调方法>>>回调消息ID为: 0a6e426a-c93a-4f53-9f1d-0375b8f61a41
4. 2019-07-07 11:22:27.762  INFO 12528 --- [ 127.0.0.1:5672] c.w.s.s.c.CustomConfirmAndReturnCallback : confirm回调方法>>>消息发送到交换机失败！，原因 : [channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'message_confirm_exchange2' in vhost '/vhost', class-id=60, method-id=40)]
5. 2019-07-07 11:22:27.762  INFO 12528 --- [ 127.0.0.1:5672] c.w.s.s.c.CustomConfirmAndReturnCallback : confirm回调方法>>>回调消息ID为: d11ad471-3a4c-4bfe-804b-eda966c4df51
6. 2019-07-07 11:22:27.762  INFO 12528 --- [ 127.0.0.1:5672] c.w.s.s.c.CustomConfirmAndReturnCallback : confirm回调方法>>>消息发送到交换机失败！，原因 : [channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'message_confirm_exchange2' in vhost '/vhost', class-id=60, method-id=40)]