顺序消息的概念与业务场景

什么是顺序消息顺序消息是指消息的发送顺序和消费顺序保持一致。比如你发了消息 A、B、C那么 Consumer 收到和处理消息的顺序也一定是 A、B、C。典型业务场景场景 为什么需要顺序订单状态流转 创建 → 支付 → 发货 → 完成必须按顺序执行绝不能先完成再创建库存扣减日志 同一商品的入库和出库操作必须按时间顺序处理否则库存数据会错乱数据库 binlog 同步 数据库变更日志必须按顺序回放否则数据不一致IoT 设备指令 对同一设备的控制指令必须有序执行如开 → 关 → 开✅ 顺序消费创建订单支付发货完成严格按照业务顺序执行→ 数据正确❌ 乱序消费创建订单发货支付支付可能在创建之前被消费→ 数据错乱全局顺序消息与分区顺序消息RocketMQ 的顺序消息分为两个级别全局顺序消息一个 Topic 下只有一个 MessageQueue所有消息严格按顺序发送和消费代价吞吐量极低单队列串行无法并行消息1唯一的 Queue消息2消息3Consumer串行消费全局有序 1 个 Topic 只有 1 个 Queue性能瓶颈明显一般不建议分区顺序消息Topic 下有多个 Queue但同一个业务 Key 的消息始终落到同一个 Queue不同 Key 的消息可以在不同 Queue 中并行互不影响这是 RocketMQ 的默认推荐方式既保证了顺序又保持了并行度多队列并行业务 Key 分组哈希取模哈希取模哈希取模订单 A: 创建→支付→发货订单 B: 创建→支付→发货订单 C: 创建→支付→发货Queue 0串行消费订单 AQueue 1串行消费订单 BQueue 2串行消费订单 C同一个订单的消息在同一个 Queue 中严格有序不同订单在不同的 Queue 中并行消费顺序消息的发送原理队列选择器 MessageQueueSelector顺序消息的发送关键在生产者端——必须保证相同业务 Key 的消息被分配到同一个 Queue。RocketMQ 提供了 MessageQueueSelector 接口让你自定义选择逻辑// 顺序消息发送示例producer.send(msg, new MessageQueueSelector() {Overridepublic MessageQueue select(List mqs, Message msg, Object arg) {// arg 是业务 Key比如订单 IDString orderId (String) arg;// 对订单 ID 取模确保同一个订单始终落在同一个 Queueint index Math.abs(orderId.hashCode()) % mqs.size();return mqs.get(index);}}, orderId);结果order_123: Queue 0order_678: Queue 1order_123 再次发送: Queue 0队列选择器工作流程消息携带业务 Key如 orderId 12345计算 Key 的哈希值对可用 Queue 数量取模返回固定的 Queue相同 Key → 相同 Queue不同 Key → 可能不同 Queue关键点如果某个 Broker 宕机该 Broker 上的 Queue 会从可用列表中移除此时哈希取模的结果会变化可能导致顺序被打破。RocketMQ 的故障规避机制在这种情况下会特殊处理尽量保持顺序但极端情况下仍可能乱序。所以顺序消息的 Broker 高可用很重要。顺序消息的消费原理顺序消费线程模型顺序消息在消费端通过 MessageListenerOrderly 实现其核心是每个 Queue 由单线程串行消费消费线程池队列锁拉取层拉取拉取拉取PullMessageServiceQueue 0Queue 1Queue 2锁定 Queue 0锁定 Queue 1锁定 Queue 2线程 1串行处理 Queue 0 的消息线程 2串行处理 Queue 1 的消息线程 3串行处理 Queue 2 的消息每个 Queue 在同一时刻只被一个线程处理不同 Queue 之间并行互不影响核心机制队列锁Consumer 在拉取消息时会对 Queue 加锁rebalanceLock确保同一个 Queue 在同一时刻只被一个线程处理串行消费一个 Queue 中的消息按拉取顺序逐个调用 MessageListenerOrderly 处理全局锁如果某个 Consumer 宕机Rebalance 会将该 Queue 分配给其他实例新实例会尝试获取队列锁确保不会有两个实例同时消费同一个 Queue// 顺序消费监听器示例consumer.registerMessageListener(new MessageListenerOrderly() {Overridepublic ConsumeOrderlyStatus consumeMessage(List msgs, ConsumeOrderlyContext context) {// 消息会按顺序被调用每次一批通常是一条for (MessageExt msg : msgs) {processOrder(msg); // 业务处理}return ConsumeOrderlyStatus.SUCCESS;}});顺序消息的 Rebalance 处理顺序消息的 Rebalance 比并发消息更复杂因为涉及到队列锁的交接BrokerConsumer BConsumer ABrokerConsumer BConsumer A正常状态C1 消费 Queue 0,1C2 消费 Queue 2,3C1 开始消费 Queue 2,3但在此之前C1 已消费完 Queue 0,1 的当前批次✅ 整个过程保证了每个 Queue 同一时刻只被一个 Consumer 消费Consumer B 宕机触发 Rebalance检测到 C2 下线重新分配 Queue尝试锁定 Queue 2,3锁定成功Queue 2,3 无锁Rebalance 中的顺序保障先停止后分配在 Rebalance 过程中Consumer 会先释放自己持有的队列锁再重新分配锁定机制新分配的 Queue 需要成功锁定后才能开始消费避免两个实例同时消费同一个 Queue消息重复风险由于 Offset 提交是异步的Rebalance 时仍可能重复消费业务层依然需要幂等