面试题：消息队列之RocketMQ在分布式系统中的消息可靠性保障

消息不丢失机制

生产者环节：
- 同步发送确认：生产者采用同步发送消息的方式，如使用 send(msg) 方法，该方法会等待 RocketMQ 服务端的响应。只有当服务端成功接收并持久化消息后，才会返回成功，否则会抛出异常。例如：
```
try {
    SendResult sendResult = producer.send(message);
    if (sendResult.getSendStatus() == SendStatus.SEND_OK) {
        // 消息发送成功处理逻辑
    } else {
        // 消息发送失败处理逻辑，如重试等
    }
} catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
    // 异常处理，如重试
}
```
- 消息持久化策略：RocketMQ 支持同步刷盘和异步刷盘两种策略。同步刷盘时，消息写入到 CommitLog 文件并刷盘成功后才返回成功响应给生产者，保证消息不会因 Broker 宕机丢失。配置示例：在 broker.conf 文件中设置 flushDiskType = SYNC_FLUSH。
Broker环节：
- 多副本机制：RocketMQ 通过主从架构，采用多副本（至少一个主副本和一个从副本）的方式来存储消息。主副本接收到消息后，会同步给从副本。当主副本出现故障时，从副本可以切换为主副本继续提供服务，从而保证消息不会丢失。例如，默认情况下，Master 节点接收到消息后会同步给 Slave 节点，只有当 Master 和 Slave 都成功持久化消息后，才认为消息成功存储。
- 高可用机制：NameServer 集群负责管理 Broker 集群的元数据信息。当某个 Broker 出现故障时，NameServer 能够感知并及时通知生产者和消费者，使它们能够切换到其他可用的 Broker 继续进行消息的生产和消费，确保消息不会因 Broker 故障而丢失。

消费者环节：

手动确认机制：消费者采用手动确认消息消费的方式。例如在 RocketMQ 消费者代码中，当消费成功后手动调用 consumer.commitSync() 方法进行确认。如果消费失败，不进行确认，RocketMQ 会根据重试策略进行重试。示例代码如下：

consumer.registerMessageListener(new MessageListenerConcurrently() {
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        for (MessageExt msg : msgs) {
            try {
                // 处理消息逻辑
                // 处理成功后手动确认
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            } catch (Exception e) {
                // 消费失败处理逻辑
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
});

消息不重复消费机制

幂等性处理：
- 业务层面幂等：消费者在业务处理逻辑上保证幂等性。例如，对于订单创建业务，在消费消息创建订单前，先查询订单是否已存在，如果已存在则不重复创建。示例代码（以数据库操作举例）：
```
// 假设使用 JDBC 操作数据库
String sql = "SELECT COUNT(*) FROM orders WHERE order_id =?";
PreparedStatement pstmt = connection.prepareStatement(sql);
pstmt.setString(1, orderId);
ResultSet rs = pstmt.executeQuery();
if (rs.next() && rs.getInt(1) > 0) {
    // 订单已存在，不重复创建
    return;
}
// 执行订单创建逻辑
```
- 消息唯一标识：RocketMQ 消息本身有唯一的 MessageId。消费者可以利用这个唯一标识，如将其存储到 Redis 等缓存中，每次消费前先检查该 MessageId 是否已处理过，如果已处理则跳过。示例代码（使用 Redis 实现）：
```
Jedis jedis = new Jedis("localhost", 6379);
String messageId = msg.getMsgId();
if (jedis.setnx(messageId, "processed") == 0) {
    // 消息已处理过，跳过
    return;
}
// 处理消息逻辑
```

各个环节保障消息可靠性总结

生产者：通过同步发送确认和合理的持久化策略，确保消息成功发送到 Broker 并持久化。
Broker：利用多副本和高可用机制，保证消息在 Broker 集群中的可靠存储和服务的持续可用。
消费者：采用手动确认机制保证消息被正确消费，通过幂等性处理避免重复消费。通过这些机制的协同工作，RocketMQ 在分布式系统中能够有效地保障消息的可靠性，确保消息不丢失、不重复消费。

星途面试题库

面试题：消息队列之RocketMQ在分布式系统中的消息可靠性保障

知识考点

面试题答案

消息不丢失机制

消息不重复消费机制

各个环节保障消息可靠性总结