怎么保证RabbitMq消息不丢失

生产者

• 做消息的持久化
  
  
  
  • 消息持久化不能保证完全不丢失消息，可以存在存储磁盘的时候还没有存储完成，但是服务宕机了也会导致消息丢失，通过发布确定保证消息不丢失
    
  
  
• 消息确定机制
  

1. import com.rabbitmq.client.*;
2. public class PersistentProducer {
3.     private final static String QUEUE_NAME = "persistent_queue";
5.     public static void main(String[] argv) throws Exception {
6.         ConnectionFactory factory = new ConnectionFactory();
7.         factory.setHost("localhost");
8.         try (Connection connection = factory.newConnection();
9.              Channel channel = connection.createChannel()) {
11.             // 声明一个持久化队列
12.             channel.queueDeclare(QUEUE_NAME, true, false, false, null);
14.             // 启用生产者确认
15.             channel.confirmSelect();
17.             String message = "Persistent message with producer confirm!";
18.             channel.basicPublish("", QUEUE_NAME,
19.                 MessageProperties.PERSISTENT_TEXT_PLAIN,
20.                 message.getBytes());
22.             // 检查消息是否成功发送
23.             if (channel.waitForConfirms()) {
24.                 System.out.println("Message sent successfully!");
25.             } else {
26.                 System.out.println("Message failed to send!");
27.             }
28.         }
29.     }
30. }

复制代码

• 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）：如果消息因某些原因无法被消耗，可以将消息转发到死信队列进行进一步处理。
  

1. import com.rabbitmq.client.*;
3. public class DirectExchangeProducer {
4.     private final static String EXCHANGE_NAME = "direct_logs";
5.     private final static String QUEUE_NAME = "persistent_queue";
7.     public static void main(String[] argv) throws Exception {
8.         ConnectionFactory factory = new ConnectionFactory();
9.         factory.setHost("localhost");
10.         try (Connection connection = factory.newConnection();
11.              Channel channel = connection.createChannel()) {
13.             // 声明交换机和队列
14.             channel.exchangeDeclare(EXCHANGE_NAME, "direct");
15.             channel.queueDeclare(QUEUE_NAME, true, false, false, null);
16.             channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "info");
18.             String message = "Hello, Direct Exchange!";
19.             channel.basicPublish(EXCHANGE_NAME, "info",
20.                 MessageProperties.PERSISTENT_TEXT_PLAIN,
21.                 message.getBytes());
23.             System.out.println("Sent: " + message);
24.         }
25.     }
26. }
28. public class DeadLetterConsumer {
29.     private final static String DLX_QUEUE = "dlx_queue";
31.     public static void main(String[] argv) throws Exception {
32.         ConnectionFactory factory = new ConnectionFactory();
33.         factory.setHost("localhost");
34.         try (Connection connection = factory.newConnection();
35.              Channel channel = connection.createChannel()) {
37.             // 声明死信队列
38.             channel.queueDeclare(DLX_QUEUE, true, false, false, null);
40.             // 创建消费者回调
41.             DeliverCallback deliverCallback = (consumerTag, delivery) -> {
42.                 String message = new String(delivery.getBody(), "UTF-8");
43.                 System.out.println("Dead Letter Queue Received: " + message);
44.             };
46.             // 设置死信队列消费者
47.             channel.basicConsume(DLX_QUEUE, true, deliverCallback, consumerTag -> {});
48.         }
49.     }
50. }

复制代码

消耗者

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

1. import com.rabbitmq.client.*;
3. public class AckConsumer {
4.     private final static String QUEUE_NAME = "persistent_queue";
6.     public static void main(String[] argv) throws Exception {
7.         ConnectionFactory factory = new ConnectionFactory();
8.         factory.setHost("localhost");
9.         try (Connection connection = factory.newConnection();
10.              Channel channel = connection.createChannel()) {
12.             // 声明一个持久化队列
13.             channel.queueDeclare(QUEUE_NAME, true, false, false, null);
15.             // 创建一个消费者回调
16.             DeliverCallback deliverCallback = (consumerTag, delivery) -> {
17.                 String message = new String(delivery.getBody(), "UTF-8");
18.                 System.out.println("Received: " + message);
20.                 try {
21.                     // 模拟消息处理
22.                     if (message.contains("error")) {
23.                         throw new Exception("Error while processing message");
24.                     }
26.                     // 手动确认消息
27.                     channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
28.                     System.out.println("Message processed and acknowledged");
29.                 } catch (Exception e) {
30.                     // 如果消息处理失败，可以将消息重新放回队列
31.                     System.out.println("Error processing message, requeueing: " + e.getMessage());
32.                     channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
33.                 }
34.             };
36.             // 设置手动确认
37.             channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> {});
38.         }
39.     }
40. }

复制代码

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

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。