消息顺序性与重复消费？

# 乱序原因 # 1. 多个队列/分区并行消费 # 消息1 → 队列1 → 消费者1（处理慢） # 消息2 → 队列2 → 消费者2（处理快） # 消息2 先被处理 → 乱序 # 2. 重试导致乱序 # 消息1 处理失败 → 重试 → 延迟 # 消息2 处理成功 → 先完成 # 3. 网络延迟 # 消息1 网络延迟 → 晚到达 # 消息2 先到达 → 先处理

# 全局有序：所有消息发到同一个队列 # 缺点：并行度低，性能差 # 适用：对顺序要求极高、消息量小的场景 # 分区有序：同一业务 key 的消息发到同一队列 # 优点：不同 key 的消息并行处理，性能好 # 适用：大部分业务场景 # 发送顺序消息 public void sendOrderly(Order order) { // 用订单 ID 作为选择队列的 key // 同一订单的消息发到同一队列 rocketMQTemplate.syncSendOrderly( "order-topic", order, order.getOrderId() // hashKey ); } # 消费顺序消息 @RocketMQMessageListener( topic = "order-topic", consumerGroup = "order-group", consumeMode = ConsumeMode.ORDERLY // 顺序消费 ) @Component public class OrderConsumer implements RocketMQListener<Order> { @Override public void onMessage(Order order) { // 同一队列的消息串行处理 processOrder(order); } }

# Kafka 只保证分区内有序 # 同一 key 的消息发到同一分区 producer.send(new ProducerRecord<>("topic", order.getOrderId(), order)); # 消费端 # 一个分区只能被一个消费者线程处理 # 多线程消费需要保证分区内串行

# 重复原因1：生产端重试 # 生产者发送消息 → 网络超时 → 重试 # Broker 实际已收到第一条 → 两条相同消息 # 重复原因2：消费端 Rebalance # 消费者 A 正在处理消息 → 触发 Rebalance # 分区重新分配给消费者 B → B 从上次 offset 开始消费 # A 处理的消息被 B 再次处理 # 重复原因3：消费超时重试 # 消费者处理消息超时 → Broker 认为失败 # 消息重新投递 → 重复消费 # 重复原因4：手动提交 offset 失败 # 消费者处理完消息 → 提交 offset 失败 # 下次拉取 → 从旧 offset 开始 → 重复消费

# RocketMQ 消费重试 # 默认重试 16 次 # 重试间隔：10s, 30s, 1m, 2m, 3m, 4m, 5m, 6m, 7m, 8m, 9m, 10m, 20m, 30m, 1h, 2h # 16 次后仍失败 → 进入死信队列（DLQ） # 死信队列 # 主题名：%DLQ%consumerGroup # 需要人工处理死信消息 # 可以重新投递或丢弃 # 消费重试配置 @RocketMQMessageListener( topic = "order-topic", consumerGroup = "order-group", maxReconsumeTimes = 5 // 最大重试次数 ) # 自定义重试逻辑 @Override public void onMessage(Order order) { try { processOrder(order); } catch (BusinessException e) { // 业务异常不重试，直接记录 log.error("业务异常，不重试", e); } catch (Exception e) { // 系统异常，重试 throw new RuntimeException("系统异常，重试", e); } }

# 场景：订单状态变更 # 创建 → 支付 → 发货 → 完成 # 必须按顺序处理，不能先发货后支付 # 方案：分区有序消息 # 同一订单的消息发到同一队列 rocketMQTemplate.syncSendOrderly("order-status-topic", msg, orderId); # 消费端顺序消费 @RocketMQMessageListener( topic = "order-status-topic", consumerGroup = "order-status-group", consumeMode = ConsumeMode.ORDERLY ) # 业务层也做状态校验 public void processOrder(Order order) { Order dbOrder = orderDao.selectById(order.getId()); // 状态校验：不能跳过状态 if (!isValidTransition(dbOrder.getStatus(), order.getStatus())) { log.warn("非法状态转换: {} → {}", dbOrder.getStatus(), order.getStatus()); return; } orderDao.updateStatus(order.getId(), order.getStatus()); }