RabbitMQ中的消息确认机制是什么？为什么需要消息确认？

向右看齐 2024-03-03 01:51 52阅读 0赞

## RabbitMQ中的消息确认机制是什么？为什么需要消息确认？ ##

RabbitMQ中的消息确认机制是指生产者发送消息后，等待消费者确认消息已经被正确接收和处理的一种机制。消息确认机制的主要目的是确保消息的可靠传递和处理，以避免消息丢失或重复处理的情况发生。

为什么需要消息确认机制呢？在分布式系统中，消息的发送和接收是异步的过程，可能会存在以下情况：

1.  消息丢失：在消息发送过程中，可能由于网络故障、硬件故障或其他原因导致消息丢失。如果没有消息确认机制，生产者无法得知消息是否成功传递给消费者，从而无法保证消息的可靠性。
2.  消息重复：在消息发送过程中，可能由于网络超时、消费者故障或其他原因导致消息重复发送。如果没有消息确认机制，消费者可能会多次处理同一条消息，导致重复操作和数据不一致的问题。

为了解决以上问题，RabbitMQ引入了消息确认机制。消息确认机制包括两个重要的概念：发布确认（Publish Confirm）和消费确认（Consumer Acknowledgement）。

发布确认是指生产者发送消息后，等待RabbitMQ服务器返回确认消息的过程。生产者可以通过调用`channel.confirmSelect()`方法启用发布确认模式，然后使用`channel.waitForConfirms()`方法等待服务器返回确认消息。如果服务器成功接收到消息并进行处理，会返回一个确认消息给生产者。

下面是一个使用Java编写的代码案例，演示了如何使用消息确认机制：

import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.ConfirmListener;
    
    import java.io.IOException;
    import java.util.Collections;
    import java.util.SortedSet;
    import java.util.TreeSet;
    import java.util.concurrent.TimeoutException;
    
    public class MessageConfirmationExample {
        
    
        private final static String QUEUE_NAME = "my_queue";
        private final static String MESSAGE = "Hello, RabbitMQ!";
    
        public static void main(String[] args) throws IOException, TimeoutException {
        
            // 创建连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("localhost");
    
            // 创建连接
            Connection connection = factory.newConnection();
    
            // 创建通道
            Channel channel = connection.createChannel();
    
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);
    
            // 启用发布确认模式
            channel.confirmSelect();
    
            // 创建一个有序集合，用于保存未确认的消息的Delivery Tag
            SortedSet<Long> unconfirmedSet = Collections.synchronizedSortedSet(new TreeSet<>());
    
            // 添加发布确认监听器
            channel.addConfirmListener(new ConfirmListener() {
        
                @Override
                public void handleAck(long deliveryTag, boolean multiple) throws IOException {
        
                    if (multiple) {
        
                        // 多条消息被确认，移除所有小于等于deliveryTag的消息
                        unconfirmedSet.headSet(deliveryTag + 1).clear();
                    } else {
        
                        // 单条消息被确认，移除该消息
                        unconfirmedSet.remove(deliveryTag);
                    }
                    System.out.println("Message confirmed: " + deliveryTag);
                }
    
                @Override
                public void handleNack(long deliveryTag, boolean multiple) throws IOException {
        
                    if (multiple) {
        
                        // 多条消息未被确认，重新发送所有小于等于deliveryTag的消息
                        unconfirmedSet.headSet(deliveryTag + 1).forEach(tag -> {
        
                            try {
        
                                sendMessage(channel, tag);
                            } catch (IOException e) {
        
                                e.printStackTrace();
                            }
                        });
                    } else {
        
                        // 单条消息未被确认，重新发送该消息
                        sendMessage(channel, deliveryTag);
                    }
                    System.out.println("Message not confirmed: " + deliveryTag);
                }
            });
    
            // 发送消息
            for (int i = 1; i <= 10; i++) {
        
                long deliveryTag = sendMessage(channel, i);
                unconfirmedSet.add(deliveryTag);
            }
    
            // 等待所有消息被确认
            try {
        
                channel.waitForConfirmsOrDie();
                System.out.println("All messages confirmed!");
            } catch (InterruptedException e) {
        
                e.printStackTrace();
            }
    
            // 关闭通道和连接
            channel.close();
            connection.close();
        }
    
        private static long sendMessage(Channel channel, int messageNumber) throws IOException {
        
            long deliveryTag = channel.getNextPublishSeqNo();
            channel.basicPublish("", QUEUE_NAME, null, (MESSAGE + " " + messageNumber).getBytes());
            System.out.println("Sent message: " + messageNumber);
            return deliveryTag;
        }
    }

在上面的代码中，首先我们创建了一个连接工厂，并设置RabbitMQ服务器的主机地址。然后，我们使用连接工厂创建了一个连接，并使用连接创建了一个通道。接下来，我们声明了一个名为"my\_queue"的队列。然后，我们启用了发布确认模式，通过调用`channel.confirmSelect()`方法。同时，我们创建了一个有序集合`unconfirmedSet`，用于保存未确认的消息的Delivery Tag。

然后，我们添加了一个发布确认监听器`ConfirmListener`，在监听器中实现了`handleAck`和`handleNack`方法。当消息被确认时，`handleAck`方法会被调用，我们可以在该方法中处理确认的逻辑，例如从`unconfirmedSet`中移除已确认的消息。当消息未被确认时，`handleNack`方法会被调用，我们可以在该方法中处理未确认的逻辑，例如重新发送未确认的消息。

接着，我们使用`sendMessage`方法发送了10条消息，并将每条消息的Delivery Tag保存到`unconfirmedSet`中。然后，我们使用`channel.waitForConfirmsOrDie()`方法等待所有消息被确认。如果在指定的时间内没有收到所有消息的确认消息，会抛出`InterruptedException`异常。