面试题：在 RabbitMQ 中，如何确保消息不会丢失？

在 RabbitMQ 中，确保消息不会丢失是一个关键问题，尤其是在处理关键业务（如支付、订单处理等）时。为了实现消息的可靠性传递，可以从以下几个方面入手：

1. 生产者端：确保消息发送成功

生产者需要采取措施确保消息成功发送到 RabbitMQ。

（1）开启事务或使用 Publisher Confirms

事务机制：

使用事务可以确保消息被成功写入 RabbitMQ。

示例代码（Java）：


channel.txSelect(); // 开启事务
try {
    channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
    channel.txCommit(); // 提交事务
} catch (Exception e) {
    channel.txRollback(); // 回滚事务
}
channel.txSelect(); // 开启事务
try {
    channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
    channel.txCommit(); // 提交事务
} catch (Exception e) {
    channel.txRollback(); // 回滚事务
}
channel.txSelect(); // 开启事务
try {
    channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
    channel.txCommit(); // 提交事务
} catch (Exception e) {
    channel.txRollback(); // 回滚事务
}

缺点：事务会显著降低性能。

Publisher Confirms（推荐）：

Publisher Confirms 是一种轻量级的确认机制，RabbitMQ 会通知生产者消息是否成功到达。

示例代码（Java）：


channel.confirmSelect(); // 开启 Publisher Confirms
channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
if (channel.waitForConfirms()) {
    System.out.println("Message confirmed");
} else {
    System.out.println("Message lost");
}
channel.confirmSelect(); // 开启 Publisher Confirms
channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
if (channel.waitForConfirms()) {
    System.out.println("Message confirmed");
} else {
    System.out.println("Message lost");
}
channel.confirmSelect(); // 开启 Publisher Confirms
channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
if (channel.waitForConfirms()) {
    System.out.println("Message confirmed");
} else {
    System.out.println("Message lost");
}

（2）消息持久化

生产者发送消息时，需将消息标记为持久化（delivery_mode=2 或 MessageProperties.PERSISTENT_TEXT_PLAIN），确保消息被写入磁盘。
示例代码（Java）：channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());

2. 队列和交换机：确保持久化

队列持久化：
- 声明队列时设置为持久化（durable=true），确保队列在 RabbitMQ 服务重启后仍然存在。
- 示例代码（Java）：channel.queueDeclare(QUEUE_NAME, true, false, false, null);
交换机持久化：
- 如果使用了自定义交换机，也需要将其声明为持久化。
- 示例代码（Java）：channel.exchangeDeclare(EXCHANGE_NAME, "direct", true);

3. 消费者端：确保消息被正确处理

消费者需要采取措施确保消息被正确处理，并防止因异常导致消息丢失。

（1）手动确认机制（ACK/NACK）

默认情况下，RabbitMQ 在消息分发给消费者后会立即删除消息。如果消费者在处理消息时崩溃，消息会丢失。
使用手动确认机制（autoAck=false），只有当消费者显式发送 ACK 时，RabbitMQ 才会删除消息。

示例代码（Java）：


channel.basicConsume(QUEUE_NAME, false, (consumerTag, delivery) -> {
    String message = new String(delivery.getBody(), "UTF-8");
    try {
        // 处理消息
        System.out.println("Received: " + message);
        // 确认消息
        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    } catch (Exception e) {
        // 消息处理失败，可以选择重新入队或丢弃
        channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
    }
}, consumerTag -> {});
channel.basicConsume(QUEUE_NAME, false, (consumerTag, delivery) -> {
    String message = new String(delivery.getBody(), "UTF-8");
    try {
        // 处理消息
        System.out.println("Received: " + message);
        // 确认消息
        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    } catch (Exception e) {
        // 消息处理失败，可以选择重新入队或丢弃
        channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
    }
}, consumerTag -> {});
channel.basicConsume(QUEUE_NAME, false, (consumerTag, delivery) -> {
    String message = new String(delivery.getBody(), "UTF-8");
    try {
        // 处理消息
        System.out.println("Received: " + message);
        // 确认消息
        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    } catch (Exception e) {
        // 消息处理失败，可以选择重新入队或丢弃
        channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
    }
}, consumerTag -> {});

（2）幂等性设计

消费者的处理逻辑应设计为幂等的，即使消息被重复消费也不会对系统造成影响。
例如，通过唯一消息 ID 或数据库唯一约束来避免重复处理。

4. 集群和高可用性

镜像队列：
- 在 RabbitMQ 集群中配置镜像队列，确保消息在多个节点之间同步。
- 如果主节点宕机，其他节点可以继续提供服务。
- 示例命令：rabbitmqctl set_policy ha-all "^ha\." '{"ha-mode":"all"}'
死信队列（DLX）：
- 配置死信队列，捕获无法正常处理的消息（如 TTL 超时、被拒绝的消息），以便后续分析和重试。