rabbitmq 怎么避免消息丢失？

避免 RabbitMQ 消息丢失需从生产者、Broker、消费者三方面构建可靠机制，通过确认机制、持久化、镜像队列等手段保障消息传递的可靠性。

• 开启 Publisher Confirm 机制：在 RabbitTemplate 中设置 publisher-confirm-type 为 correlated（推荐）或 simple。correlated 模式通过维护消息序号与回调函数的映射，提供更详细的确认信息；simple 模式仅通过返回值判断是否成功。示例代码：

  @Beanpublic RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) { RabbitTemplate template = new RabbitTemplate(connectionFactory); template.setPublisherConfirmType(RabbitTemplate.ConfirmType.CORRELATED); return template;}

• 处理 Confirm 回调：实现 CorrelationData.ConfirmCallback 接口，在回调方法中根据 ack 状态处理成功或失败逻辑。若消息发送失败，需重发或记录日志以便排查。示例代码：

  @Componentpublic 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: 重发消息或记录日志 } }}

• 消息持久化：

  Exchange 和 Queue 持久化：在声明时设置 durable=true，确保 Broker 重启后元数据不丢失。示例代码：@Beanpublic Queue myQueue() { return new Queue("myQueue", true); // true 表示持久化}@Beanpublic DirectExchange myExchange() { return new DirectExchange("myExchange", true, false); // true 表示持久化}

  Message 持久化：发送消息时设置 MessageProperties.deliveryMode 为 PERSISTENT，确保消息以持久化方式存储。示例代码：MessageProperties messageProperties = new MessageProperties();messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);Message message = new Message("Hello, RabbitMQ!".getBytes(), messageProperties);rabbitTemplate.convertAndSend("myExchange", "myRoutingKey", message);

• 镜像队列：通过策略配置实现队列数据在集群节点间的同步备份。当主节点故障时，镜像节点自动接管，保证消息可靠性。配置命令：

  rabbitmqctl set_policy ha-all "^" '{"ha-mode":"all"}'

  此命令将所有队列设置为镜像队列，所有节点同步队列数据。

• 开启手动 ACK 机制：设置 spring.rabbitmq.listener.acknowledge-mode=manual，消费者需手动发送 ACK 确认消息已被正确消费。示例配置：

  spring.rabbitmq.listener.acknowledge-mode=manual

• 处理 ACK 与异常：

  消息处理成功后，调用 channel.basicAck() 发送 ACK。

  处理失败时，调用 channel.basicNack() 或 channel.basicReject() 拒绝消息，并根据需求设置 requeue=true（重新入队）或 false（进入死信队列）。示例代码：

  @RabbitListener(queues = "myQueue")public void processMessage(Message message, Channel channel) throws IOException { try { System.out.println("Received message: " + new String(message.getBody())); channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); } catch (Exception e) { channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true); // 或直接丢弃消息：channel.basicReject(message.getMessageProperties().getDeliveryTag(), false); }}

• 死信队列作用：存储因拒绝、过期或队列满而无法消费的消息，便于后续处理（如记录日志、人工干预或重发）。
• 配置步骤：

  创建队列时指定 x-dead-letter-exchange 和 x-dead-letter-routing-key 参数。示例代码：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。示例代码：@Beanpublic DirectExchange deadLetterExchange() { return new DirectExchange("deadLetterExchange");}@Beanpublic Queue deadLetterQueue() { return new Queue("deadLetterQueue");}@Beanpublic Binding deadLetterBinding() { return BindingBuilder.bind(deadLetterQueue()).to(deadLetterExchange()).with("deadLetterRoutingKey");}

• 生产者端：

  检查 Publisher Confirm 机制是否开启及回调处理是否正确。

  确认消息是否成功发送至 Broker（通过抓包工具或管理界面）。

  检查 mandatory 参数是否为 true（确保无法路由的消息返回错误）。

• Broker 端：

  验证 Exchange、Queue 和 Message 的持久化配置。

  检查镜像队列策略是否生效。

  查看 Broker 日志排查异常。

• 消费者端：

  确认手动 ACK 机制是否开启及处理逻辑是否正确。

  检查消费者是否成功消费消息或发生异常导致消息丢失。

通过以上策略的综合应用，可有效避免 RabbitMQ 消息丢失，构建高可靠性的消息传递系统。