RokcetMq的顺序消费、防丢失、去重 - 文章 - 开发者社区

RocketMQ 作为阿里巴巴开源的分布式消息中间件，经历了多年双11大促的考验，在顺序消息、消息不丢失、消息防重复三大核心能力上有着成熟的设计。下面深入解析其实现原理。

一、顺序消息

1.1 什么是顺序消息？

顺序消息是 RocketMQ 提供的一种高级消息类型，支持消费者按照发送消息的先后顺序获取消息。它基于消息组（MessageGroup） 来判定和识别顺序关系，相同消息组的消息遵循先进先出（FIFO）原则。

1.2 顺序消息的应用场景

顺序消息适用于对消息顺序有严格要求的业务场景：

场景	说明
订单处理	同一订单的创建→支付→发货→完成必须按顺序处理，否则状态会紊乱
撮合交易	证券/股票交易中，价格相同时先出价者优先成交
数据库变更同步	MySQL binlog 同步时，操作顺序必须与源库一致
日志同步	有序事件处理，下游需要按时间顺序还原数据

1.3 顺序消息的实现原理

RocketMQ 的顺序性保障分为生产顺序性和消费顺序性两个环节。

生产顺序性

要保证消息生产的顺序性，必须满足两个条件：

单一生产者：不同生产者之间无法保证消息顺序，因为分布式环境下无法判定先后
串行发送：多线程并行发送会导致顺序混乱，必须单线程串行发送

生产端代码示例：

java

ini
 体验AI代码助手
 代码解读
复制代码
public class OrderProducer {
    public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("order_producer_group");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();
        
        String[] tags = new String[]{"create", "pay", "ship", "finish"};
        
        for (int i = 0; i < 100; i++) {
            int orderId = i % 10;  // 假设有10个订单
            Message msg = new Message("OrderTopic", tags[i % tags.length], 
                ("order data " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
            
            // 关键：使用 MessageQueueSelector 将同一订单的消息路由到同一队列
            SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                @Override
                public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                    Integer id = (Integer) arg;
                    int index = id % mqs.size();  // 按订单ID取模，保证同一订单进同一队列
                    return mqs.get(index);
                }
            }, orderId);
            
            System.out.printf("Send result: %s%n", sendResult);
        }
        
        producer.shutdown();
    }
}

消费顺序性

消费端需要确保消息按照存储顺序被处理，RocketMQ 通过 MessageListenerOrderly 实现队列内串行消费。

消费端代码示例：

java

java
 体验AI代码助手
 代码解读
复制代码
public class OrderConsumer {
    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("order_consumer_group");
        consumer.setNamesrvAddr("localhost:9876");
        consumer.subscribe("OrderTopic", "*");
        
        // 使用 MessageListenerOrderly 保证顺序消费
        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, 
                                                        ConsumeOrderlyContext context) {
                context.setAutoCommit(true);  // 自动提交 offset
                for (MessageExt msg : msgs) {
                    System.out.printf("Thread: %s, QueueId: %d, MsgId: %s, Content: %s%n",
                        Thread.currentThread().getName(),
                        msg.getQueueId(),
                        msg.getMsgId(),
                        new String(msg.getBody()));
                    
                    // 业务处理逻辑
                    processOrder(msg);
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        
        consumer.start();
        System.out.println("Order Consumer Started.");
    }
}

1.4 顺序消息的注意事项

主题类型：顺序消息只能发送到类型为 FIFO 的主题
性能权衡：顺序消费会限制并发度，建议使用分区顺序而非全局顺序
故障处理：消费失败时返回 SUSPEND_CURRENT_QUEUE_A_MOMENT，暂停该队列消费，避免后续消息处理导致状态不一致
高可用说明：严格顺序模式下，Broker 宕机会影响顺序性保障，需权衡可用性与顺序性

二、消息不丢失机制

RocketMQ 通过生产者、Broker、消费者三端的协同设计，实现消息全链路不丢失。

2.1 消息丢失的可能环节

环节	风险点
生产者 → Broker	网络闪断、发送超时
Broker 存储	磁盘故障、进程崩溃
Broker → 消费者	消费异常、确认丢失

2.2 生产者端保障

同步发送 + 重试机制是最可靠的发送方式：

java

ini
 体验AI代码助手
 代码解读
复制代码
DefaultMQProducer producer = new DefaultMQProducer("producer_group");
producer.setNamesrvAddr("localhost:9876");
producer.setRetryTimesWhenSendFailed(3);  // 设置重试次数
producer.setSendMsgTimeout(5000);          // 设置超时时间
producer.start();

Message msg = new Message("TopicTest", "TagA", "Hello RocketMQ".getBytes());

// 同步发送，阻塞等待 Broker 确认
SendResult sendResult = producer.send(msg);
if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
    // 发送失败，业务兜底：记录到数据库，定时重试
    saveToFailQueue(msg);
}

关键配置：

同步发送确保收到 Broker ACK 才返回
设置重试次数，应对临时故障
发送失败需业务兜底（持久化后定时重发）

2.3 Broker 端保障

同步刷盘（SYNC_FLUSH）

同步刷盘确保消息写入物理磁盘后才返回 ACK，即使 Broker 宕机也不会丢消息。

java

arduino
 体验AI代码助手
 代码解读
复制代码
// Broker 配置
flushDiskType = SYNC_FLUSH  // 同步刷盘

刷盘模式	说明	可靠性	性能
SYNC_FLUSH	消息落盘后才返回 ACK	高（不丢失）	较低
ASYNC_FLUSH	写入缓冲区即返回，定时刷盘	中（可能丢失）	高

主从复制（SYNC_MASTER）

同步复制模式下，主节点等待从节点存储成功后才返回确认。

java

arduino
 体验AI代码助手
 代码解读
复制代码
// Broker 配置
brokerRole = SYNC_MASTER  // 同步复制

复制模式对比：

复制模式	说明	可靠性
SYNC_MASTER	主从都写入成功才返回	高，主宕机从仍可服务
ASYNC_MASTER	主写入即返回，异步复制到从	较低，主宕机可能丢消息

2.4 消费者端保障

手动 ACK是消费端防止丢失的关键：

java

kotlin
 体验AI代码助手
 代码解读
复制代码
consumer.registerMessageListener((msgs, context) -> {
    for (MessageExt msg : msgs) {
        try {
            // 业务处理
            processMessage(msg);
        } catch (Exception e) {
            // 消费失败，触发重试
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    }
    // 全部成功才确认
    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});

重试机制：

消费失败返回 RECONSUME_LATER，Broker 会重新投递
默认重试 16 次
16 次均失败后进入死信队列（DLQ） ，需人工介入处理

三、消息防重复机制

RocketMQ 采用 "至少一次（At Least Once）" 投递语义，可能造成消息重复，需通过幂等设计解决。

3.1 重复消息的产生场景

生产者重试导致重复发送
消费者 ACK 超时，消息重新投递
Broker 主从切换导致重复消费

3.2 消息去重方案

方案一：基于业务 Key 的幂等

不要依赖 Message ID，不同消息可能内容相同。正确做法是使用业务唯一标识作为 Key：

java

ini
 体验AI代码助手
 代码解读
复制代码
// 生产者设置业务 Key
Message message = new Message();
message.setKeys("ORDER_12345");  // 使用订单号作为唯一标识
SendResult sendResult = producer.send(message);

方案二：数据库唯一约束

利用数据库唯一索引天然幂等：

java

typescript
 体验AI代码助手
 代码解读
复制代码
public void consumeOrder(MessageExt msg) {
    String orderId = msg.getKeys();
    try {
        orderDao.insert(order);  // 订单表 order_id 唯一索引
    } catch (DuplicateKeyException e) {
        log.warn("订单已存在，跳过重复消费: {}", orderId);
    }
}

方案三：Redis 去重

使用 Redis 的 setnx 原子操作，设置合理过期时间：

java

typescript
 体验AI代码助手
 代码解读
复制代码
public boolean consumeWithRedis(MessageExt msg) {
    String msgId = msg.getMsgId();
    // 设置 5 分钟过期，setnx 成功说明首次处理
    Boolean success = redisTemplate.opsForValue()
        .setIfAbsent("msg:" + msgId, "1", Duration.ofMinutes(5));
    
    if (Boolean.TRUE.equals(success)) {
        // 处理业务
        processMessage(msg);
        return true;
    }
    // 已处理过，直接跳过
    log.info("消息已处理过，跳过: {}", msgId);
    return false;
}

方案四：状态机设计

java

scss
 体验AI代码助手
 代码解读
复制代码
// 订单状态：0-待支付，1-已支付，2-已发货，3-已完成
public void payOrder(String orderId) {
    Order order = orderDao.select(orderId);
    if (order.getStatus() != 0) {
        log.warn("订单状态不是待支付，忽略支付请求: status={}", order.getStatus());
        return;
    }
    // 更新状态为已支付
    orderDao.updateStatus(orderId, 1);
}

四、三大机制总结

机制	核心目标	关键技术	配置要点
顺序消费	保证消息按发送顺序处理	MessageQueueSelector + MessageListenerOrderly	主题需为 FIFO 类型，相同业务 Key 路由到同一队列
消息不丢失	全链路可靠性保障	同步发送 + 同步刷盘 + 同步复制 + 手动 ACK	`flushDiskType=SYNC_FLUSH` `brokerRole=SYNC_MASTER`
消息防重复	幂等消费	业务唯一 Key + 去重表/唯一约束 + 状态机	生产端设置 `setKeys`，消费端幂等设计

五、最佳实践建议

5.1 生产环境推荐配置

properties

ini
 体验AI代码助手
 代码解读
复制代码
# Broker 端（追求高可靠性）
flushDiskType = SYNC_FLUSH
brokerRole = SYNC_MASTER
# 生产者端
producer.setRetryTimesWhenSendFailed(3)
producer.setSendMsgTimeout(5000)
# 消费者端
consumer.setConsumeMessageBatchMaxSize(1)  # 顺序消费建议单条

5.2 性能与可靠性权衡

业务场景	顺序性要求	可靠性要求	推荐配置
订单核心链路	高	高	同步发送 + SYNC_FLUSH + SYNC_MASTER + 手动 ACK
日志采集	低	中	异步发送 + ASYNC_FLUSH + ASYNC_MASTER
支付通知	中	高	同步发送 + SYNC_FLUSH + 业务幂等兜底

5.3 监控与治理

部署 RocketMQ Console 监控消息积压
关注死信队列，及时人工处理
设置告警：消息积压量、消费延迟、发送失败率

RocketMQ 通过这三套机制的协同保障，在生产环境中能够达到 99.99% 以上的可靠性，这也是它能经受住双十一考验的重要原因。