避免消息丢失需从生产者、Broker、消费者三方面入手：生产者启用Publisher Confirm机制并处理回调；Broker端对Exchange、Queue、Message进行持久化，并配置镜像队列；消费者开启手动ACK，确保消息处理成功后确认。同时配置死信队列处理异常消息，结合监控排查问题，保障消息可靠传递。
rabbitmq 避免消息丢失的核心在于从生产者到 Broker，Broker 自身，以及 Broker 到消费者的整个链路中，都要采取相应的策略来保证消息的可靠传递。这需要生产者确认机制、持久化设置、消费者确认机制以及镜像队列等多种手段的配合。
解决方案
避免 RabbitMQ 消息丢失需要从生产者、Broker 和消费者三个方面入手，构建一个完整的可靠消息传递机制。
生产者端：开启 Publisher Confirm 机制
生产者是消息的起点，确保消息成功发送到 RabbitMQ Broker 至关重要。Publisher Confirm 机制允许 Broker 在收到消息后向生产者发送确认，告知消息已被正确接收。
-
开启 Publisher Confirm: 在 RabbitTemplate 中设置
publisher-confirm-type为correlated或simple。correlated模式需要生产者维护一个消息序号与回调函数的映射，simple模式则只能通过返回值来判断是否成功。推荐使用correlated模式，因为它能提供更详细的确认信息。 -
处理 Confirm 回调: 实现
ConfirmCallback接口，在回调方法中处理 Broker 返回的确认信息。如果消息被成功接收，则执行相应的业务逻辑；如果消息发送失败，则需要进行重发或者记录日志，方便后续排查问题。
@Component public class MyConfirmCallback implements CorrelationData.ConfirmCallback { @Override public void confirm(CorrelationData correlationData, boolean ack, String cause) { if (ack) { System.out.println("消息发送成功，correlationData: " + correlationData); } else { System.out.println("消息发送失败，correlationData: " + correlationData + ", cause: " + cause); // TODO: 重发消息或记录日志 } } } @Autowired private RabbitTemplate rabbitTemplate; @Autowired private MyConfirmCallback myConfirmCallback; public void sendMessage(String message) { CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString()); rabbitTemplate.setConfirmCallback(myConfirmCallback); rabbitTemplate.convertAndSend("myExchange", "myRoutingKey", message, correlationData); }
Broker 端：消息持久化与镜像队列
Broker 作为消息的中转站，需要保证消息在 Broker 内部的可靠存储，以及 Broker 集群间的数据同步。
- 消息持久化: 将 Exchange、Queue 和 Message 都设置为持久化。Exchange 和 Queue 的持久化确保 Broker 重启后，这些元数据不会丢失。Message 的持久化确保消息在 Broker 内部以持久化的方式存储，即使 Broker 发生故障，消息也不会丢失。
@Bean public Queue myQueue() { return new Queue("myQueue", true); // true 表示持久化 } @Bean public DirectExchange myExchange() { return new DirectExchange("myExchange", true, false); // true 表示持久化 }
在发送消息时，需要设置 MessageProperties 的 deliveryMode 为 PERSISTENT。
MessageProperties messageProperties = new MessageProperties(); messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT); Message message = new Message("Hello, RabbitMQ!".getBytes(), messageProperties); rabbitTemplate.convertAndSend("myExchange", "myRoutingKey", message);
- 镜像队列: 使用镜像队列可以实现 Broker 集群中队列数据的同步备份。当主节点发生故障时，镜像节点可以自动接管，保证消息的可靠性。镜像队列可以通过 RabbitMQ 的策略配置来实现。
rabbitmqctl set_policy ha-all "^" '{"ha-mode":"all"}'
这个命令会将所有队列设置为镜像队列，所有节点都会同步队列的数据。
消费者端：开启 ACK 机制
消费者是消息的终点，需要确保消息被正确消费，并且在消费失败时能够进行重试。
-
开启 ACK 机制: 将
correlated0 设置为correlated1。这意味着消费者需要手动发送 ACK 确认消息已被正确消费。 -
处理 ACK: 在消息处理成功后，调用
correlated2 方法发送 ACK。如果消息处理失败，可以调用correlated3 或correlated4 方法拒绝消息，并将消息重新放入队列，以便后续重试。
@RabbitListener(queues = "myQueue") public void processMessage(Message message, channel channel) throws IOException { try { // 处理消息 System.out.println("Received message: " + new String(message.getBody())); // 发送 ACK channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); } catch (Exception e) { // 处理失败，重新放入队列 channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true); // 或者直接丢弃消息 // channel.basicReject(message.getMessageProperties().getDeliveryTag(), false); } }
消息丢失了，如何排查问题？
排查消息丢失问题需要从生产者、Broker 和消费者三个方面入手，逐一排除可能的原因。
-
生产者端:
-
Broker 端:
- 检查 Exchange 和 Queue 是否设置为持久化。
- 检查消息是否设置为持久化。
- 检查镜像队列是否正确配置。
- 检查 Broker 的日志，查看是否有异常信息。
-
消费者端:
- 检查 ACK 机制是否正确开启和处理。
- 检查消费者是否成功消费消息。
- 检查消费者是否在处理消息时发生异常，导致消息丢失。
如何保证消息的顺序性？
在某些场景下，消息的顺序性非常重要。RabbitMQ 默认情况下并不能保证消息的绝对顺序性，但可以通过一些策略来尽量保证消息的顺序性。
-
单一队列: 将所有需要保证顺序性的消息发送到同一个队列。这样可以保证消息在队列中的顺序与发送的顺序一致。
-
单一消费者: 使用单一消费者来消费队列中的消息。这样可以避免多个消费者并发消费导致的消息顺序错乱。
-
消息分片: 将消息按照一定的规则进行分片，然后将每个分片发送到不同的队列。每个队列使用单一消费者来消费，保证每个分片内部的顺序性。然后将所有分片按照顺序进行组装，得到完整的消息。
如何处理死信队列？
死信队列（Dead Letter Queue，DLQ）用于存储无法被正常消费的消息。这些消息通常是因为以下原因被放入死信队列：
- 消息被拒绝（
correlated8 或correlated9）且simple0 参数设置为simple1。 - 消息过期。
- 队列达到最大长度。
通过配置死信队列，可以方便地对无法被正常消费的消息进行处理，例如：
- 记录日志: 将死信队列中的消息记录到日志中，方便后续排查问题。
- 人工干预: 将死信队列中的消息发送给人工处理。
- 重发消息: 将死信队列中的消息重新发送到队列中，进行重试。
配置死信队列需要在创建队列时指定 simple2 和 simple3 参数。
Map<String, Object> args = new HashMap<>(); args.put("x-dead-letter-exchange", "deadLetterExchange"); args.put("x-dead-letter-routing-key", "deadLetterRoutingKey"); Queue queue = new Queue("myQueue", true, false, false, args);
然后创建一个 Exchange 和 Queue 用于存储死信消息。
@Bean public DirectExchange deadLetterExchange() { return new DirectExchange("deadLetterExchange"); } @Bean public Queue deadLetterQueue() { return new Queue("deadLetterQueue"); } @Bean public Binding deadLetterBinding() { return BindingBuilder.bind(deadLetterQueue()).to(deadLetterExchange()).with("deadLetterRoutingKey"); }
总而言之，避免 RabbitMQ 消息丢失是一个系统工程，需要从生产者、Broker 和消费者三个方面进行综合考虑，采取相应的策略来保证消息的可靠传递。同时，需要建立完善的监控和告警机制，及时发现和处理潜在的问题。
评论（已关闭）
评论已关闭