怎么保证RabbitMq消息不丢失

生产者

• 做消息的持久化
• 消息确定机制

import com.rabbitmq.client.*;
public class PersistentProducer {private final static String QUEUE_NAME = "persistent_queue";public static void main(String[] argv) 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);// 启用生产者确认channel.confirmSelect();String message = "Persistent message with producer confirm!";channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());// 检查消息是否成功发送if (channel.waitForConfirms()) {System.out.println("Message sent successfully!");} else {System.out.println("Message failed to send!");}}}
}

• channel.queueDeclare(QUEUE_NAME, true, false, false, null)：队列是持久化的，确保 RabbitMQ 重启后队列不会丢失。
• MessageProperties.PERSISTENT_TEXT_PLAIN：确保消息持久化存储。
• channel.confirmSelect()：启用生产者确认，确保消息成功送达 RabbitMQ。
• channel.waitForConfirms()：等待确认，如果生产者发送消息时发生失败，会捕获错误。

交换机

• 选择合适的交换机类型：常见的交换机类型包括 direct、fanout、topic 和 headers，选择正确的类型来确保消息路由正确。
• 使用死信队列（Dead Letter Exchange, DLX）：如果消息因某些原因无法被消费，可以将消息转发到死信队列进行进一步处理。

import com.rabbitmq.client.*;public class DirectExchangeProducer {private final static String EXCHANGE_NAME = "direct_logs";private final static String QUEUE_NAME = "persistent_queue";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {// 声明交换机和队列channel.exchangeDeclare(EXCHANGE_NAME, "direct");channel.queueDeclare(QUEUE_NAME, true, false, false, null);channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "info");String message = "Hello, Direct Exchange!";channel.basicPublish(EXCHANGE_NAME, "info", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());System.out.println("Sent: " + message);}}
}public class DeadLetterConsumer {private final static String DLX_QUEUE = "dlx_queue";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {// 声明死信队列channel.queueDeclare(DLX_QUEUE, true, false, false, null);// 创建消费者回调DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println("Dead Letter Queue Received: " + message);};// 设置死信队列消费者channel.basicConsume(DLX_QUEUE, true, deliverCallback, consumerTag -> {});}}
}

消费者

• 进行手动应答
• 消息重试

import com.rabbitmq.client.*;public class AckConsumer {private final static String QUEUE_NAME = "persistent_queue";public static void main(String[] argv) 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);// 创建一个消费者回调DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println("Received: " + message);try {// 模拟消息处理if (message.contains("error")) {throw new Exception("Error while processing message");}// 手动确认消息channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);System.out.println("Message processed and acknowledged");} catch (Exception e) {// 如果消息处理失败，可以将消息重新放回队列System.out.println("Error processing message, requeueing: " + e.getMessage());channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);}};// 设置手动确认channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> {});}}
}

• channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false)：消息处理成功后确认消息。
• channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true)：如果消费失败，重新将消息投递到队列中，供其他消费者处理。