如何玩懂RabbitMQ的消息确认机制？

📢📢📢📣📣📣

哈喽！大家好，我是【一心同学】，一位上进心十足的【Java领域博主】！😜😜😜

✨【一心同学】的写作风格：喜欢用【通俗易懂】的文笔去讲解每一个知识点，而不喜欢用【高大上】的官方陈述。

✨【一心同学】博客的领域是【面向后端技术】的学习，未来会持续更新更多的【后端技术】以及【学习心得】。

✨如果有对【后端技术】感兴趣的【小可爱】，欢迎关注【一心同学】💞💞💞

❤️❤️❤️感谢各位大可爱小可爱！❤️❤️❤️

问题引入

我们在使用RabbitMQ的时候，我们可以通过消息持久化操作来解决因为服务器的异常崩溃导致的消息丢失，除此之外我们还会遇到一个问题，当消息的发布者在将消息发送出去之后，消息到底有没有正确到达broker代理服务器呢？

如果不进行特殊配置的话，默认情况下发布操作是不会返回任何信息给生产者的，也就是默认情况下我们的生产者是不知道消息有没有正确到达broker的，如果在消息到达broker之前已经丢失的话，持久化操作也解决不了这个问题，因为消息根本就没到达代理服务器，我们是无法进行持久化的。

消费者收到消息后的处理也是如此，例如还没来得及"消费"它,或者说还没来得及进行业务逻辑处理时,消费者所在的信道或者连接因某种原因断开了，那这条消息岂不是就被无情的抛弃了...而这个问题的解决就需要我们的消息确认机制来进行处理了。

一、生产者的消息确认

1.1 准备工作

（1）创建一个SpringBoot项目

（2）导入依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
 
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>

（3）编写配置文件

application.yml:

server:
  port: 8021
spring:
#给项目起个名字
  application:
    name: rabbitmq-provider
  #配置rabbitMq 服务器
  rabbitmq:
    host: 服务器地址
    port: 5672
    username: yixin
    password: 123456
    #虚拟host 可以不设置,使用server默认host
    virtual-host: /
    #消息确认配置项

    #确认消息已发送到交换机(Exchange)
    publisher-confirm-type: correlated
    #确认消息已发送到队列(Queue)
    publisher-returns: true

（4）编写配置相关的消息确认回调函数

package com.yixin.config;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitConfig {

    @Bean
    public RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory){
        RabbitTemplate rabbitTemplate = new RabbitTemplate();
        rabbitTemplate.setConnectionFactory(connectionFactory);
        //设置开启Mandatory,才能触发回调函数,无论消息推送结果怎么样都强制调用回调函数
        rabbitTemplate.setMandatory(true);

        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                System.out.println("ConfirmCallback:     "+"相关数据："+correlationData);
                System.out.println("ConfirmCallback:     "+"确认情况："+ack);
                System.out.println("ConfirmCallback:     "+"原因："+cause);
            }
        });

        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                System.out.println("ReturnCallback:     "+"消息："+message);
                System.out.println("ReturnCallback:     "+"回应码："+replyCode);
                System.out.println("ReturnCallback:     "+"回应信息："+replyText);
                System.out.println("ReturnCallback:     "+"交换机："+exchange);
                System.out.println("ReturnCallback:     "+"路由键："+routingKey);
            }
        });

        return rabbitTemplate;
    }

}

此时我们的生产者推送消息的消息确认调用回调函数已经编写完毕。

1.2 回调函数触发规则

可以看到我们在配置类中写了两个回调函数，一个叫 ConfirmCallback ，一个叫 RetrunCallback；
那么以上这两种回调函数都是在什么情况会触发呢？

我们从总体的情况分析，推送消息存在四种情况：

①消息推送到server，但是在server里找不到交换机
②消息推送到server，找到交换机了，但是没找到队列
③消息推送到sever，交换机和队列啥都没找到
④消息推送成功

接下来我们进行编写几个接口来分别测试和认证以上4种情况，看看消息确认触发回调函数的触发情况：

①消息推送到server，但是在server里找不到交换机

注：​non-existent-exchange 这个交换机是不存在的。

package com.yixin.controller;

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

@RestController
public class FannoutController {

    @Autowired
    RabbitTemplate rabbitTemplate;  //使用RabbitTemplate,这提供了接收/发送等等方法

    @GetMapping("/TestMessageAck")
    public String TestMessageAck() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "message: non-existent-exchange test message ";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String, Object> map = new HashMap<>();
        map.put("messageId", messageId);
        map.put("messageData", messageData);
        map.put("createTime", createTime);
        
        //non-existent-exchange这个交换机是不存在的。
       rabbitTemplate.convertAndSend("lonelyFanoutExchange", null,map);
        return "ok";
    }
}

浏览器输入：http://localhost:8021/TestMessageAck 进行调用接口，查看rabbitmq-provider项目的控制台输出情况:

ConfirmCallback:     相关数据：null
ConfirmCallback:     确认情况：false
ConfirmCallback:     原因：channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'non-existent-exchange' in vhost '/', class-id=60, method-id=40)

** 原因中表示没有找到交换机'non-existent-exchange'。**

结论： ①这种情况触发的是 ConfirmCallback 回调函数。

** ②消息推送到server，找到交换机了，但是没找到队列 **

（1）我们需要新增一个交换机，但是不给这个交换机绑定队列，我来简单地在FannoutRabbitMQ里面新增一个Fanout交换机，名叫‘lonelyFanouttExchange’，但没给它做任何绑定配置操作：

package com.yixin.config;

import org.springframework.amqp.core.FanoutExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class FannoutRabbitMQ {

    @Bean
    FanoutExchange lonelyFanoutExchange() {
        return new FanoutExchange("lonelyFanoutExchange");
    }
}

（2）编写写个测试接口，把消息推送到名为‘lonelyFanoutExchange’的交换机上（这个交换机是没有任何队列配置的）：

    @GetMapping("/TestMessageAck2")
    public String TestMessageAck2() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "message: lonelyDirectExchange test message ";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String, Object> map = new HashMap<>();
        map.put("messageId", messageId);
        map.put("messageData", messageData);
        map.put("createTime", createTime);
        
       rabbitTemplate.convertAndSend("lonelyFanoutExchange", null,map);
        return "ok";
    }

调用接口，查看rabbitmq-provider项目的控制台输出情况：

ReturnCallback:

ReturnCallback:     消息：(Body:'[serialized object]' MessageProperties [headers={}, contentType=application/x-java-serialized-object, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, deliveryTag=0])
ReturnCallback:     回应码：312
ReturnCallback:     回应信息：NO_ROUTE
ReturnCallback:     交换机：lonelyFanoutExchange
ReturnCallback:     路由键：TestFanoutRouting

ConfirmCallback:

ConfirmCallback:     相关数据：null
ConfirmCallback:     确认情况：true
ConfirmCallback:     原因：null

分析：

（1）两个函数都被调用了。
（2）消息是推送成功到服务器了的，所以ConfirmCallback对消息确认情况是true；
（3）在RetrunCallback回调函数的打印参数里面可以看到，消息是推送到了交换机成功了，但是在路由分发给队列的时候，找不到队列，所以报了错误 NO_ROUTE 。

** 结论：②这种情况触发的是 ConfirmCallback和RetrunCallback两个回调函数。**

**③消息推送到sever，交换机和队列啥都没找到 **

这种情况一看就觉得跟①很像，没错 ，③和①情况回调是一致的，所以不做结果说明了。

结论： ③这种情况触发的是 ConfirmCallback 回调函数。

④消息推送成功

（1）配置中绑定队列和交换机

package com.yixin.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class FannoutRabbitMQ {

    @Bean
    public Queue queueConfirm() {
        return new Queue("fanout.confirm",true);
    }

    @Bean
    FanoutExchange confirmFanouttExchange() {
        return new FanoutExchange("lonelyFanoutExchange");
    }
    @Bean
    Binding bindingExchangeConfirm() {
        return BindingBuilder.bind(queueConfirm()).to(confirmFanouttExchange());
    }

}

（2）编写接口进行调用

    @GetMapping("/TestMessageAck4")
    public String TestMessageAck4() {

        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "message: lonelyDirectExchange test message ";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String, Object> map = new HashMap<>();
        map.put("messageId", messageId);
        map.put("messageData", messageData);
        map.put("createTime", createTime);

   rabbitTemplate.convertAndSend("lonelyFanoutExchange", null,map);
        return "ok";
    }

调用接口进行测试，控制台如下：

ConfirmCallback:     相关数据：null
ConfirmCallback:     确认情况：true
ConfirmCallback:     原因：null

结论： ④这种情况触发的是 ConfirmCallback 回调函数。

1.3 小结

①消息推送到server，但是在server里找不到交换机
触发 ConfirmCallback 回调函数。
②消息推送到server，找到交换机了，但是没找到队列
触发ConfirmCallback和ReturnCallback回调函数
③消息推送到sever，交换机和队列啥都没找到。
触发 ConfirmCallback 回调函数。
④消息推送成功
触发 ConfirmCallback 回调函数。

二、消费者的消息确认

消费者的消息确认机制不同，因为消息接收本来就是在监听消息，符合条件的消息就会消费下来。
所以，消息接收的确认机制主要存在两种种模式：

2.1 自动确认

这也是默认的消息确认情况。在自动确认模式中,消息在发送到消费者后即被认为"成功消费"。这种模式可以降低吞吐量（只要消费者可以跟上）,以降低交付和消费者处理的安全性。

这种模式通常被称为“即发即忘”。与手动确认模型不同,如果消费者的TCP连接或通道在真正的"成功消费"之前关闭,则服务器发送的消息将丢失.因此,自动消息确认应被视为不安全,并不适用于所有工作负载。

2.2 手动确认

这个比较关键，也是我们配置接收消息确认机制时，多数选择的模式。
消费者收到消息后，手动调用basic.ack / basic.nack / basic.reject后，RabbitMQ收到这些消息后，才认为本次投递成功。

2.2.1 确认的方法

**basic.ack用于肯定确认
basic.nack用于否定确认（注意：这是AMQP 0-9-1的RabbitMQ扩展）
basic.reject用于否定确认，但与basic.nack相比有一个限制:一次只能拒绝单条消息 **

注：

消费者端以上的3个方法都表示消息已经被正确投递

basic.ack：表示消息已经被正确处理。

basic.nack和basic.reject表示没有被正确处理。

2.2.2 basic.nack

channel.basicNack(deliveryTag, false, true);

第一个参数：依然是当前消息到的数据的唯一id;
第二个参数：指是否针对多条消息；如果是true，也就是说一次性针对当前通道的消息的tagID小于当前这条消息的，都拒绝确认。
第三个参数：是指是否重新入列，也就是指不确认的消息是否重新丢回到队列里面去。

同样使用不确认后重新入列这个确认模式要谨慎，因为这里也可能因为考虑不周出现消息一直被重新丢回去的情况，导致积压。

2.2.3 basic.reject

channel.basicReject(deliveryTag, true);  

拒绝消费当前消息，如果第二参数传入true，就是将数据重新丢回队列里，那么下次还会消费这消息。设置false，就是告诉服务器，我已经知道这条消息数据了，因为一些原因拒绝它，而且服务器也把这个消息丢掉就行, 下次不想再消费这条消息了。

注意：

（1）使用拒绝后重新入列这个确认模式要谨慎，因为一般都是出现异常的时候，catch异常再拒绝入列，选择是否重入列。

（2）如果使用不当会导致一些每次都被你重入列的消息一直消费-入列-消费-入列这样循环，会导致消息积压。

2.3 手动确认的配置

2.3.1 监听单个队列

（1）创建一个SpringBoot的消费者项目

（2）导入依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
 
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>

（3）编写配置文件

server:
  port: 8022
spring:
  #给项目起个名字
  application:
    name: rabbitmq-consumer
  #配置rabbitMq 服务器
  rabbitmq:
    host: 服务器地址
    port: 5672
    username: yixin
    password: 123456
    #虚拟host 可以不设置,使用server默认host
    virtual-host: /

（4）编写配置类

package com.yixin.config;

import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class MessageListenerConfig {

    @Autowired
    private CachingConnectionFactory connectionFactory;
    @Autowired
    private MyAckReceiver myAckReceiver;//消息接收处理类

    @Bean
    public SimpleMessageListenerContainer simpleMessageListenerContainer() {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        container.setConcurrentConsumers(1);
        container.setMaxConcurrentConsumers(1);
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL); // RabbitMQ默认是自动确认，这里改为手动确认消息
        //设置一个队列
        container.setQueueNames("fanout.confirm");
        //如果同时设置多个如下： 前提是队列都是必须已经创建存在的
        //  container.setQueueNames("TestDirectQueue","TestDirectQueue2","TestDirectQueue3");

        //另一种设置队列的方法,如果使用这种情况,那么要设置多个,就使用addQueues
        //container.setQueues(new Queue("TestDirectQueue",true));
        //container.addQueues(new Queue("TestDirectQueue2",true));
        //container.addQueues(new Queue("TestDirectQueue3",true));
        container.setMessageListener(myAckReceiver);

        return container;
    }

}

（5）编写消息监听类

我们需要编写对应的手动确认消息监听类，MyAckReceiver.java（手动确认模式需要实现 ChannelAwareMessageListener）：

package com.yixin.config;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;

@Component
public class MyAckReceiver implements ChannelAwareMessageListener {

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
           
            String msg = message.toString();

            System.out.println("消费者收到消息:"+msg);

            System.out.println("消费的主题消息来自："+message.getMessageProperties().getConsumerQueue());
            channel.basicAck(deliveryTag, true); //第二个参数，手动确认可以被批处理，当该参数为 true 时，则可以一次性确认 delivery_tag 小于等于传入值的所有消息
//            channel.basicReject(deliveryTag, true);//第二个参数，true会重新放回队列，所以需要自己根据业务逻辑判断什么时候使用拒绝
        } catch (Exception e) {
            channel.basicReject(deliveryTag, false);
            e.printStackTrace();
        }
    }

}

（6）编写调用接口

在我们的生产者那边编写消息的投递。

    @GetMapping("/TestMessageAck5")
    public String TestMessageAck5() {

        String msg="我是一心同学";
        rabbitTemplate.convertAndSend("lonelyFanoutExchange", null,msg);
        return "发送成功 ";
    }

进行调用，投递消息，查看我们的消费者控制台：

** 可以看到监听器正常消费了下来。如果我们在消息监听类中注释掉channel.basicAck，那么就是是消息被消费者拿到了，仍旧会存在队列中的。**

2.3.2 监听多个队列

场景描述：监听的多个队列并变成手动确认模式，而且没个队列处理的消息业务逻辑不一样。

我们在生产者那边多绑定一个队列，其余不变：

package com.yixin.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class FannoutRabbitMQ {

    @Bean
    public Queue queueConfirm() {
        return new Queue("fanout.confirm",true);
    }

    @Bean
    public Queue queueConfirmMore() {
        return new Queue("fanout.confirmMore",true);
    }

    @Bean
    FanoutExchange confirmFanouttExchange() {
        return new FanoutExchange("lonelyFanoutExchange");
    }
    @Bean
    Binding bindingExchangeConfirm() {
        return BindingBuilder.bind(queueConfirm()).to(confirmFanouttExchange());
    }

    @Bean
    Binding bindingExchangeConfirmMore() {
        return BindingBuilder.bind(queueConfirmMore()).to(confirmFanouttExchange());
    }

}

现在我们交换机已经绑定了fanout.confirm和fanout.confirmMore两个队列。（我们也可以多注册几个交换机，每个交换机绑定各自的队列，这里只是为了方便而已）

消费者配置

配置类：

package com.yixin.config;

import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class MessageListenerConfig {

    @Autowired
    private CachingConnectionFactory connectionFactory;
    @Autowired
    private MyAckReceiver myAckReceiver;//消息接收处理类

    @Bean
    public SimpleMessageListenerContainer simpleMessageListenerContainer() {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        container.setConcurrentConsumers(1);
        container.setMaxConcurrentConsumers(1);
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL); // RabbitMQ默认是自动确认，这里改为手动确认消息
        //设置一个队列
      //  container.setQueueNames("fanout.confirm");
        //如果同时设置多个如下： 前提是队列都是必须已经创建存在的
         container.setQueueNames("fanout.confirm","fanout.confirmMore");

        //另一种设置队列的方法,如果使用这种情况,那么要设置多个,就使用addQueues
        //container.setQueues(new Queue("TestDirectQueue",true));
        //container.addQueues(new Queue("TestDirectQueue2",true));
        //container.addQueues(new Queue("TestDirectQueue3",true));
        container.setMessageListener(myAckReceiver);

        return container;
    }

}

消息监听类

手动确认消息监听类，**MyAckReceiver.java **就可以同时将上面设置到的队列的消息都消费下来并且我们需要做不用的业务逻辑处理，那么只需要 根据消息来自的队列名进行区分处理即可：

package com.yixin.config;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;

@Component
public class MyAckReceiver implements ChannelAwareMessageListener {

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
         
            String msg = message.toString();

            
            if ("fanout.confirm".equals(message.getMessageProperties().getConsumerQueue())) {
                System.out.println("消费的消息来自的队列名为：" + message.getMessageProperties().getConsumerQueue());
                System.out.println("消费者收到消息:"+msg);
                System.out.println("执行fanout.confirm中的消息的业务处理流程......");
            }
            if ("fanout.confirmMore".equals(message.getMessageProperties().getConsumerQueue())){
                System.out.println("消费的消息来自的队列名为："+message.getMessageProperties().getConsumerQueue());
                System.out.println("消费者收到消息:"+msg);
                System.out.println("执行fanout.confirmMore中的消息的业务处理流程......");

            }

            channel.basicAck(deliveryTag, true); //第二个参数，手动确认可以被批处理，当该参数为 true 时，则可以一次性确认 delivery_tag 小于等于传入值的所有消息
//            channel.basicReject(deliveryTag, true);//第二个参数，true会重新放回队列，所以需要自己根据业务逻辑判断什么时候使用拒绝
        } catch (Exception e) {
            channel.basicReject(deliveryTag, false);
            e.printStackTrace();
        }
    }

}

现在我们调用生产者那边的接口进行投递消息，查看消费者这边的控制台：

** 可以看到我们的我们不同的队列都被监听了，并且执行了各自的逻辑。**

小结

以上就是【一心同学】整理的【RabbitMQ】的【消息确认机制】，我们可以利用【消息确认机制】进而避免我们的消息丢失，从而【保护数据】。

如果这篇【文章】有帮助到你，希望可以给【一心同学】点个赞👍，创作不易，相比官方的陈述，我更喜欢用【通俗易懂】的文笔去讲解每一个知识点，如果有对【后端技术】感兴趣的小可爱，也欢迎关注❤️❤️❤️ 【一心同学】❤️❤️❤️，我将会给你带来巨大的【收获与惊喜】💕💕！