⭐⭐⭐ Spring Boot 项目实战	⭐⭐⭐ Spring Cloud 项目实战
《Dubbo 实现原理与源码解析 —— 精品合集》	《Netty 实现原理与源码解析 —— 精品合集》
《Spring 实现原理与源码解析 —— 精品合集》	《MyBatis 实现原理与源码解析 —— 精品合集》
《Spring MVC 实现原理与源码解析 —— 精品合集》	《数据库实体设计合集》
《Spring Boot 实现原理与源码解析 —— 精品合集》	《Java 面试题 + Java 学习指南》

摘要: 原创出处民工哥技术之路「民工哥技术之路」欢迎转载，保留摘要，谢谢！

🙂🙂🙂关注微信公众号：【芋道源码】有福利：

RocketMQ / MyCAT / Sharding-JDBC 所有源码分析文章列表

RocketMQ / MyCAT / Sharding-JDBC 中文注释源码 GitHub 地址

您对于源码的疑问每条留言都将得到认真回复。甚至不知道如何读源码也可以请教噢。

新的源码解析文章实时收到通知。每周更新一篇左右。

认真的源码交流微信群。

前面介绍了 RabbitMQ 安装与简单使用、基础配置、管理界面介绍、常用 API、消息收发与交换机等相关的知识点，今天我将详细的为大家介绍 RabbitMQ 消息确认机制相关知识，希望大家能够从中收获多多！如有帮助，请点在看、转发支持一波！！！

RabbitMQ 消息确认机制

RabbitMQ消息确认机制指的是在消息传递过程中，发送方发送消息后，接收方需要对消息进行确认，以确保消息被正确地接收和处理。RabbitMQ的消息确认机制分为两种：

生产者确认机制：生产者发送消息后，需要等待RabbitMQ服务器的确认消息，以确保消息已经被成功地发送到RabbitMQ服务器。如果RabbitMQ服务器没有收到消息或者消息发送失败，生产者会收到一个确认消息，从而可以进行重发或者其他处理。
消费者确认机制：消费者接收到消息后，需要向RabbitMQ服务器发送确认消息，以告诉服务器已经成功地接收并处理了该消息。如果消费者没有发送确认消息，RabbitMQ服务器会认为该消息没有被正确地处理，从而会将该消息重新发送给其他消费者进行处理。

在RabbitMQ中，消息确认机制是通过ACK机制来实现的。ACK代表Acknowledgement，即确认消息。当消息发送方发送消息后，接收方需要向消息发送方发送ACK消息，以表示已经成功地接收和处理了该消息。如果消息发送方没有收到ACK消息，就会认为该消息没有被正确地处理，从而进行重发或者其他处理。

总之，RabbitMQ的消息确认机制可以保证消息的可靠性，从而提高系统的稳定性和可靠性。

消息可靠抵达 ConfirmCallback

RabbitMQ的消息确认机制确保了消息的可靠抵达，其中ConfirmCallback是其中一种实现方式。

ConfirmCallback是一个回调函数，用于在消息被确认时进行回调，以确保消息已经被正确地发送到RabbitMQ Broker并被处理。当生产者发送消息时，可以通过调用channel的confirmSelect()方法将channel设置为confirm模式，然后通过添加Confirm Callback回调函数来处理消息确认。

当消息被发送到Broker后，如果Broker成功地将消息路由到目标队列，则会调用ConfirmCallback回调函数的handleAck()方法，表示消息已被确认。如果Broker无法将消息路由到目标队列，则会调用handleNack()方法，表示消息未被确认。

使用ConfirmCallback可以确保消息已经被正确地发送到RabbitMQ Broker并被处理，从而避免了消息丢失或重复发送的情况。同时，ConfirmCallback还可以在消息未被确认时进行重试或记录日志等操作，以确保消息的可靠性和稳定性。

ConfirmCallback使用说明：

在配置文件中配置：spring.rabbitmq.publisher-confirms=true
在创建 connectionFactory 的时候设置 PublisherConfirms(true) 选项，开启 confirmcallback。

CorrelationData：用来表示当前消息唯一性。

消息只要被 broker 接收到就会执行 confirmCallback，如果是 cluster 模式，需要所有broker 接收到才会调用 confirmCallback。
被 broker 接收到只能表示 message 已经到达服务器，并不能保证消息一定会被投递到目标 queue 里。所以需要用到接下来的 returnCallback 。

消息可靠抵达-ReturnCallback

RabbitMQ的ReturnCallback机制是为了解决消息无法路由到指定队列的问题。当发送的消息无法被路由到指定队列时，RabbitMQ会将消息返回给生产者，这时候如果生产者设置了ReturnCallback回调函数，就可以在回调函数中处理这种情况。

ReturnCallback机制的使用场景一般是在消息发送时，指定了mandatory参数为true，表示如果消息无法被路由到指定队列，则将消息返回给生产者。如果mandatory参数为false，则消息会被直接丢弃。

当生产者设置了ReturnCallback回调函数后，RabbitMQ在将消息返回给生产者时，会触发该回调函数。在ReturnCallback回调函数中，可以处理消息无法路由的情况，例如重发消息、记录日志等。

下面是一个使用ReturnCallback机制的示例代码：

channel.addReturnListener(new ReturnCallback() {
    @Override
    public void handle(ReturnedMessage returnedMessage) {
        String message = new String(returnedMessage.getBody());
        System.out.println("Message returned: " + message);
    }
});

channel.basicPublish(exchangeName, routingKey, true, null, message.getBytes());

在上面的代码中，我们通过addReturnListener方法设置了ReturnCallback回调函数，当消息无法路由到指定队列时，会触发该回调函数。在回调函数中，我们将返回的消息打印出来，以便处理。

需要注意的是，ReturnCallback机制只有在消息被发送到交换机后，才会触发。如果消息发送的交换机不存在，或者路由键不符合任何绑定规则，消息会被直接丢弃，不会触发ReturnCallback回调函数。

在配置文件中配置：

spring.rabbitmq.publisher-returns=true
spring.rabbitmq.template.mandatory=true

confrim 模式只能保证消息到达 broker，不能保证消息准确投递到目标 queue 里。在有些业务场景下，我们需要保证消息一定要投递到目标 queue 里，此时就需要用到return 退回模式。

这样如果未能投递到目标 queue 里将调用 returnCallback ，可以记录下详细到投递数据，定期的巡检或者自动纠错都需要这些数据。

RabbitMQ自动确认和手动确认

RabbitMQ消息确认机制是一种保证消息可靠抵达的机制。在RabbitMQ中，消息确认机制分为两种模式：自动确认模式和手动确认模式。

在自动确认模式下，当消费者收到消息并将其处理完毕后，RabbitMQ会自动将该消息标记为已确认，然后将其从队列中删除。这种模式比较简单，但是存在消息丢失的风险，因为如果消费者在处理消息的过程中出现异常，消息就会被丢失。

在手动确认模式下，当消费者收到消息并将其处理完毕后，需要向RabbitMQ发送一个确认消息，告诉RabbitMQ该消息已经被处理完毕，可以从队列中删除。如果消费者在处理消息的过程中出现异常，可以选择不发送确认消息，这样消息就不会被从队列中删除，可以重新被其他消费者获取。

手动确认模式可以保证消息不会被丢失，但是需要消费者编写额外的代码来处理确认消息的发送。此外，手动确认模式还可以设置确认模式为批量确认模式，即一次性确认多个消息，可以提高消息处理的效率。

总结

自动ACK模式

在自动ACK模式下，消息一旦被消费者接收到，就会自动被确认。这种模式下，消息一旦被发送到消费者，就会从队列中删除，无论消费者是否已经成功消费该消息。

手动ACK模式

在手动ACK模式下，消费者必须手动发送ACK消息来确认消息已经被成功消费。如果消费者没有发送ACK消息，RabbitMQ服务器就会认为该消息还没有被消费，会将该消息重新发送给其他消费者。这种模式下，消费者可以选择性地确认消息，只有当消费者成功消费一条消息并确认后，该消息才会从队列中删除。

手动ACK模式可以保证消息的可靠性，但是需要消费者在处理消息时进行额外的处理，消费者需要在处理消息后发送ACK消息，否则消息会被重新发送。

在 RabbitMQ 中，消费者可以通过设置 channel.basicAck(deliveryTag, multiple) 方法来发送 ack 消息。其中，deliveryTag 表示消息的唯一标识符，multiple 表示是否批量确认。如果 multiple 为 true，表示要确认该 deliveryTag 及其之前的所有消息；如果 multiple 为 false，表示只确认该 deliveryTag 指定的一条消息。

需要注意的是，如果消费者在处理消息时发生了异常，消息并没有被成功处理，那么不应该发送 ack 消息，而应该将消息重新放回队列中，让其他消费者来处理。

RabbitMQ处理消息方法

RabbitMQ提供了一些基本的方法来处理消息，这些方法包括：

basic.publish: 发布消息到指定的交换机上。
basic.consume: 消费消息，启动一个消费者来监听指定队列上的消息。
basic.ack: 确认消息已经被消费，告诉RabbitMQ可以删除该消息。
basic.nack: 否认消息已经被消费，告诉RabbitMQ需要重新发送该消息。
basic.reject: 拒绝消息，告诉RabbitMQ不需要再次发送该消息。
basic.get: 获取指定队列上的一条消息。
basic.cancel: 取消消费者的消费，停止监听指定队列上的消息。

这些方法都是基于AMQP协议定义的，可以使用RabbitMQ提供的客户端库或者自己实现AMQP协议来调用这些方法。

发布确认原理

生产者将信道设置为 confirm 模式，一旦信道进入 confirm 模式，所有在该信道上发布的消息都将被指派一个唯一的 ID （从1开始），一旦消息被投递到所有匹配的队列之后，broker 就会发送一个确认个生产者（包含消息的唯一 ID），这使得生产者知道消息已经确认到达目的队列了，如果消息和队列是可持久化的，那么确认消息会在将消息写入磁盘之后发出，broker 回传给生产者的确认消息中 delivery-tag 域包含了确认消息的序列号，此外 broker 也可以设置 basic.ack 的 multiple 域，表示到这个序列号之前的所有消息都已经得到了处理。

confirm 模式最大的好处在于它是异步的，一旦发布一条消息，生产者应用程序就可以在等信道返回确认的同时继续发送下一条消息，当消息最终得到确认之后，生产者应用便可以通过回调方法来处理该确认消息，如果 RabbitMQ 因为自身内部错误导致消息丢失，就会发送一条 nack 消息，生产者应用程序通同样可以在回调方法中处理该 nack 消息。

发布确认的策略

开启发布确认的方法

发布确认默认是没有开启的，如果要开启需要调用方法 confirmSelect，每当你要想使用发布确认，都需要在 channel 上调用该方法。

Channel channel = connection.createChannel();
channel.confirmSelect();

单个确认发布

这是一种简单的确认方式，它是一种同步确认发布的方式，也就是发布一个消息之后只有它被确认发布，后续的消息才能继续发布，waitForConfirmsOrDie(long) 这个方法只有在消息被确认的时候才返回，如果在指定时间范围内这个消息没有被确认那么它将抛出异常。

这种确认方式有一个最大的缺点就是：发布速度特别的慢，因为如果没有确认发布的消息就会阻塞所有后续消息的发布，这种方式最提供每秒不超过数百条发布消息的吞吐量。当然对于某些应用程序来说这可能已经足够了。

private static void publishMessageIndividually() throws Exception {
        try (Channel channel = RabbitMqUtils.getChannel()) {
            String queueName = UUID.randomUUID().toString();
            channel.queueDeclare(queueName, false, false, false, null);
            // 开启发布确认
            channel.confirmSelect();
            long begin = System.currentTimeMillis();
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String message = i + "";
                channel.basicPublish("", queueName, null, message.getBytes());
                boolean flag = channel.waitForConfirms();
                if (flag) {
                    System.out.println("消息发送成功");
                }

            }
            long end = System.currentTimeMillis();
            System.out.println("发布" + MESSAGE_COUNT + "个单独确认消息，耗时" + (end - begin) + "ms");
        }
    }

批量确认发布

上面那种方式非常慢，与单个等待确认消息相比，先发布一批消息然后一起确认可以极大地提高吞吐量，当然这种方式的缺点就是：当发生故障导致发布出现问题时，不知道是哪个消息出现问题了，我们必须将整个批处理保存在内存中，以记录重要的信息而后重新发布消息。当然这种方案仍然是同步的，也一样阻塞消息的发布。

private static void publishMessageInBatch() throws Exception {
        try (Channel channel = RabbitMqUtils.getChannel()) {
            String queueName = UUID.randomUUID().toString();
            channel.queueDeclare(queueName, false, false, false, null);
            // 开启发布确认
            channel.confirmSelect();
            // 批量确认消息大小
            int batchSize = 100;
            // 未确认消息个数
            int outstandingMessageCount = 0;
            long begin = System.currentTimeMillis();
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String message = i + "";
                channel.basicPublish("", queueName, null, message.getBytes());
                outstandingMessageCount++;
                if (outstandingMessageCount == batchSize) {
                    channel.waitForConfirms();
                    outstandingMessageCount = 0;
                }

            }
            // 为了确保还有剩余没有确认消息 再次确认
            if(outstandingMessageCount > 0){
                channel.waitForConfirms();
            }
            long end = System.currentTimeMillis();
            System.out.println("发布" + MESSAGE_COUNT + "个批量确认消息，耗时" + (end - begin) + "ms");
        }
    }

异步确认发布

异步确认虽然编程逻辑比上两个要复杂，但是性价比最高，无论是可靠性还是效率都没得说，他是利用回调函数来达到消息可靠性传递的，这个中间件也是通过函数回调来保证是否投递成功。

public static void publishMessageAsync() throws Exception {
        try (Channel channel = RabbitMqUtils.getChannel()) {
            String queueName = UUID.randomUUID().toString();
            channel.queueDeclare(queueName, false, false, false, null);
            //开启发布确认
            channel.confirmSelect();
            /**
             * 线程安全有序的一个哈希表，适用于高并发的情况
             * 1.轻松的将序号与消息进行关联
             * 2.轻松批量删除条目 只要给到序列号
             * 3.支持并发访问
             */
            ConcurrentSkipListMap<Long, String> outstandingConfirms = new
                    ConcurrentSkipListMap<>();
            /**
             * 确认收到消息的一个回调
             * 1.消息序列号
             * 2.true 可以确认小于等于当前序列号的消息
             * false 确认当前序列号消息
             */
            ConfirmCallback ackCallback = (sequenceNumber, multiple) -> {
                if (multiple) {
                    //返回的是小于等于当前序列号的未确认消息 是一个 map
                    ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(sequenceNumber, true);
                    //清除该部分未确认消息
                    confirmed.clear();
                } else {
                    //只清除当前序列号的消息
                    outstandingConfirms.remove(sequenceNumber);
                }
            };
            ConfirmCallback nackCallback = (sequenceNumber, multiple) ->
            {
                String message = outstandingConfirms.get(sequenceNumber);
                System.out.println("发布的消息" + message + "未被确认，序列号" + sequenceNumber);
            };
            /**
             * 添加一个异步确认的监听器
             * 1.确认收到消息的回调
             * 2.未收到消息的回调
             */
            channel.addConfirmListener(ackCallback, null);
            long begin = System.currentTimeMillis();
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String message = "消息" + i;
                /**
                 * channel.getNextPublishSeqNo()获取下一个消息的序列号
                 * 通过序列号与消息体进行一个关联
                 * 全部都是未确认的消息体
                 */
                outstandingConfirms.put(channel.getNextPublishSeqNo(), message);
                channel.basicPublish("", queueName, null, message.getBytes());
            }
            long end = System.currentTimeMillis();
         System.out.println("发布" + MESSAGE_COUNT + "个异步确认消息,耗时" + (end - begin) + "ms");
        }
    }

如何处理异步未确认消息

最好的解决方案就是把未确认的消息放到一个基于内存的能被发布线程访问的队列，比如说用 ConcurrentLinkedQueue 这个队列在 confirm callbacks 与发布线程之间进行消息的传递。

以上3种确认速度对比

单独发布消息：同步等待确认，简单，但吞吐量非常有限。
批量发布消息：批量同步等待确认，简单，合理的吞吐量，一旦出现问题但很难推断出来是哪条消息出现了问题。
异步处理：最佳性能和资源使用，在出现错误的情况下可以很好地控制，但是实现起来稍微难些。

public static void main(String[] args) throws Exception {
        //这个消息数量设置为 1000 好些 不然花费时间太长
        publishMessageIndividually();
        publishMessageInBatch();
        publishMessageAsync();
}
// 运行结果
发布1000个单独确认消息，耗时549ms
发布1000个批量确认消息，耗时110ms
发布1000个异步确认消息，耗时25ms

芋道源码 —— 纯源码解析博客

硬卷消息中间件系列（七）：RabbitMQ 消息确认机制详解