3.2 消息确认机制 (Message Acknowledgements)

3.2 消息确认机制 (Message Acknowledgements)

3.2 RabbitMQ 消息确认机制 (Message Acknowledgements)

消息确认机制 (Message Acknowledgements)，简称ACK，是RabbitMQ确保消息可靠传递的关键特性。它允许消费者告知RabbitMQ服务端，消息是否已被成功处理。只有在收到ACK后，RabbitMQ才会将消息从队列中移除。如果消费者在处理消息时发生错误、崩溃或连接中断，没有发送ACK，RabbitMQ会将消息重新放入队列，以便其他消费者可以再次处理。

3.2.1 为什么需要消息确认机制

在分布式系统中，消息传递的可靠性至关重要。如果没有消息确认机制，可能会出现以下问题：

• 消息丢失： 消费者在处理消息时发生崩溃，消息未被处理就丢失了。

• 消息重复消费： 消费者在处理消息后，发送ACK之前崩溃，RabbitMQ认为消息未被处理，会将消息重新放入队列，导致消息被重复消费。

消息确认机制通过确保消息至少被成功处理一次，来解决这些问题。

3.2.2 ACK的类型

RabbitMQ支持两种类型的ACK：

• 自动确认 (Automatic Acknowledgements): 当消费者收到消息后，RabbitMQ会自动将消息标记为已确认。这种模式简单易用，但可靠性较低，适用于对消息丢失不太敏感的场景。如果消费者在处理消息时发生错误，消息可能会丢失。

• 手动确认 (Manual Acknowledgements): 消费者需要显式地发送ACK告知RabbitMQ消息已成功处理。这种模式可靠性更高，但需要消费者编写更多的代码。手动确认允许消费者在处理消息后，根据处理结果决定是否发送ACK。

  • Positive Acknowledgement (Basic.Ack): 表示消息已成功处理。

  • Negative Acknowledgement (Basic.Nack) / Basic.Reject: 表示消息处理失败，需要RabbitMQ采取相应的措施，例如重新放入队列或丢弃。

  • Basic.Nack 可以批量拒绝消息，而 Basic.Reject 只能拒绝单个消息。

3.2.3 手动确认的工作流程

1. 生产者 (Producer): 将消息发送到交换机 (Exchange)。

2. 交换机 (Exchange): 根据路由规则将消息路由到一个或多个队列 (Queue)。

3. 队列 (Queue): 存储消息，等待消费者消费。

4. 消费者 (Consumer): 从队列中获取消息。

5. 处理消息 (Process Message): 消费者处理消息。

6. 发送 ACK (Send Basic.Ack/Basic.Nack/Basic.Reject):

   • 如果消息处理成功，消费者发送 Basic.Ack 给 RabbitMQ。

   • 如果消息处理失败，消费者发送 Basic.Nack 或 Basic.Reject 给 RabbitMQ。

7. RabbitMQ:

   • 收到 Basic.Ack 后，将消息从队列中移除。

   • 收到 Basic.Nack 或 Basic.Reject 后，根据参数决定是否将消息重新放入队列。

3.2.4 代码实践 (Java)

以下代码示例展示了如何在Java中使用RabbitMQ的自动确认和手动确认机制。

1. 自动确认 (Auto Acknowledgement)

import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class AutoAckConsumer {
    private static final String QUEUE_NAME = "auto_ack_queue";
    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, false, false, false, null);
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println(" [x] Received '" + message + "'");
            // 模拟消息处理
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            System.out.println(" [x] Done");
        };
        channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { }); // autoAck = true
        // Keep the consumer running
        try {
            Thread.sleep(Long.MAX_VALUE);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        channel.close();
        connection.close();
    }
}
​

说明:

• channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { }); 中的第二个参数 true 表示启用自动确认。

• 消费者收到消息后，会自动向RabbitMQ发送ACK。

• 即使在 deliverCallback 中发生异常，消息也会被自动确认。

2. 手动确认 (Manual Acknowledgement)

import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class ManualAckConsumer {
    private static final String QUEUE_NAME = "manual_ack_queue";
    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, false, false, false, null);
        channel.basicQos(1); // 每次只接收一条消息
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println(" [x] Received '" + message + "'");
            try {
                // 模拟消息处理
                Thread.sleep(1000);
                // 假设消息处理成功
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                System.out.println(" [x] Done");
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                // 假设消息处理失败，重新放入队列
                try {
                    channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
                } catch (IOException e1) {
                    e1.printStackTrace();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        };
        channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> { }); // autoAck = false
        // Keep the consumer running
        try {
            Thread.sleep(Long.MAX_VALUE);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        channel.close();
        connection.close();
    }
}
​

说明:

• channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> { }); 中的第二个参数 false 表示禁用自动确认，启用手动确认。

• channel.basicQos(1); 设置 prefetchCount 为 1，表示消费者在收到ACK之前，RabbitMQ不会向其发送新的消息。 这可以防止消费者在处理大量消息时，因为资源耗尽而崩溃。

• channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); 发送 Basic.Ack 给 RabbitMQ，确认消息已成功处理。 delivery.getEnvelope().getDeliveryTag() 是消息的唯一标识符。 第二个参数 false 表示只确认当前消息，如果设置为 true，则会确认所有小于当前 deliveryTag 的消息。

• channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true); 发送 Basic.Nack 给 RabbitMQ，表示消息处理失败。 第二个参数 false 表示只拒绝当前消息，如果设置为 true，则会拒绝所有小于当前 deliveryTag 的消息。 第三个参数 true 表示将消息重新放入队列，false 表示丢弃消息。

• 如果消息处理过程中发生异常，我们发送 Basic.Nack 并将消息重新放入队列。

3. 使用 Basic.Reject

如果只需要拒绝单个消息，可以使用 Basic.Reject 方法：

channel.basicReject(delivery.getEnvelope().getDeliveryTag(), true); // requeue = true
​

Basic.Reject 方法与 Basic.Nack 的区别在于，Basic.Reject 只能拒绝单个消息，而 Basic.Nack 可以批量拒绝消息。

3.2.5 选择合适的ACK模式

选择哪种ACK模式取决于应用的具体需求：

• 自动确认: 适用于对消息丢失不太敏感的场景，例如日志收集。

• 手动确认: 适用于对消息可靠性要求较高的场景，例如金融交易。

在选择手动确认时，需要考虑以下因素：

• 性能: 手动确认会增加消费者的代码复杂性，并可能降低性能。

• 并发: 如果消费者是多线程的，需要确保ACK操作是线程安全的。

• 异常处理: 需要妥善处理消息处理过程中可能发生的异常，并根据情况发送 Basic.Ack 或 Basic.Nack。

3.2.6 总结

消息确认机制是RabbitMQ确保消息可靠传递的关键特性。通过选择合适的ACK模式，并编写健壮的消费者代码，可以构建可靠的消息驱动系统。手动确认模式虽然增加了复杂性，但提供了更高的可靠性，适用于对数据一致性要求严格的应用场景。在实际应用中，应根据业务需求权衡性能和可靠性，选择最合适的ACK模式。