异步解耦之RabbitMQ(四)_消息持久化及ACK机制

本文介绍: 在消息传递系统中，消息持久化是一种机制，用于保证即使在消息代理或消费者故障的情况下，消息也能够被保存并最终传递到目标。对于需要长时间存储的重要消息，消息持久化非常有用。在RabbitMQ中，可以通过设置队列的参数来启用消息持久化。当消息被标记为持久化时，它们将在磁盘上进行存储，从而确保即使在RabbitMQ服务器重启后，也能够恢复未处理的消息。本文详细介绍了RabbitMQ的消息持久化和ACK机制。

RabbitMQ是一个功能强大的开源消息代理，通过消息持久化和ACK机制，确保消息传递的可靠性。本文将详细介绍RabbitMQ的消息持久化和ACK机制。

在消息传递系统中，消息持久化是一种机制，用于保证即使在消息代理或消费者故障的情况下，消息也能够被保存并最终传递到目标。对于需要长时间存储的重要消息，消息持久化非常有用。

在RabbitMQ中，可以通过设置队列的参数来启用消息持久化。当消息被标记为持久化时，它们将在磁盘上进行存储，从而确保即使在RabbitMQ服务器重启后，也能够恢复未处理的消息。

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;

public class RabbitMQExample {
    private final static String QUEUE_NAME = "myQueue";

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");

        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {

            // 创建持久化队列
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);

            // 发送消息到队列，设置消息持久化
            String message = "Hello, RabbitMQ!";
            channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
            System.out.println("Sent message: " + message);

            // 接收消息并发送ACK
            DeliverCallback deliverCallback = (consumerTag, delivery) -> {
                String receivedMessage = new String(delivery.getBody(), "UTF-8");
                System.out.println("Received message: " + receivedMessage);
                // 模拟消息处理时间
                Thread.sleep(5000);
                // 发送ACK
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                System.out.println("ACK sent for message: " + receivedMessage);
            };

            channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> {});
        }
    }
}