MQ消息乱序问题解析与实战解决方案

作者：京东物流 刘浩

1. 背景

在分布式系统中，消息队列（MQ）是实现系统解耦、异步通信的重要工具。然而，MQ消费时出现的消息乱序问题，经常会对业务逻辑的正确执行和系统稳定性产生不良影响。本文将详细探讨MQ消息乱序问题的根源，并提供一系列在实际应用中可行的解决方案。

2. MQ消息乱序问题分析

常见的MQ消息乱序问题的根源主要可以归结为以下几点：

2.1 相同topic内的消息乱序

1). 并发消费：

在分布式系统中，为了提高消息处理的吞吐量，通常会配置多个消费者实例来并发消费同一个队列中的消息。然而，由于消费者实例的机器性能、网络延迟以及处理速度的差异，可能导致消息的消费顺序与发送顺序不一致。

2). 消息分区：

为了支持更高效的消息存储和消费，MQ系统通常会采用分区化的设计。然而，当同一业务逻辑的多条消息被分发到不同的分区时，消费者在消费这些消息时就可能出现乱序现象。

3). 网络延迟与抖动：

消息在传输过程中可能会受到网络延迟和抖动的影响，导致消息到达消费者端的时间顺序与发送顺序不一致。

4). 消息重试与故障恢复：

当消费者处理消息失败或出现故障时，MQ系统通常会进行消息重试或故障恢复操作。如果重试机制或故障恢复策略设计不当，也可能导致消息乱序。

2.2 不同topic的消息乱序

从相对时间的视角来审视，消息被消费的顺序并不等同于其被发送的顺序。例如，系统A在12:00时向TopicA发送了消息msgA-12:00，而紧接着系统B在12:01时向TopicB发送了消息msgB-12:01。当系统C同时订阅并消费这两个Topic时，它无法预设msgA-12:00会必然先于msgB-12:01被接收。这是由于消息系统在处理过程中，受到诸如消息分区策略、各个Consumer的处理能力以及其诸如网络、堆积、重试等他综合因素的影响，导致无法确保消息遵循严格的先进先出原则。

3. 案例分析

3.1 数据迁移过程中的mq消费乱序场景

在数据迁移或同步过程中，尤其是双写场景（即数据既写入旧系统，又通过MQ发送到新系统进行异步处理），MQ乱序可能导致严重的数据不一致问题。

具体来说，当数据写入时发送INSERT MQ，数据更新时发送UPDATE MQ，如果UPDATE MQ先于INSERT MQ到达目标系统，目标系统可能会基于一个不存在的数据记录进行更新操作。这会导致以下几种情况：

数据丢失：如果目标系统没有处理UPDATE MQ中提到的数据记录（因为该记录尚未通过INSERT MQ创建），则更新操作会失败，可能导致数据变更丢失或遗漏。

数据覆盖：在高频修改的情况下，频繁更新可能会面临旧数据覆盖新数据的风险，比如UPDATE MQ携带的是旧数据且先于新数据的UPDATE MQ到达。

3.2 业务风险分析

MQ乱序对数据迁移和同步过程的影响是深远的：

数据一致性受损：最直接的影响是数据一致性受损。目标系统中的数据可能与源系统不一致，导致业务决策基于错误的数据。

用户体验下降：数据不一致可能导致用户看到错误的信息或遇到功能故障，从而降低用户体验。

业务中断：在严重的情况下，数据不一致可能导致业务中断或系统故障，影响企业的运营和声誉。

4. 解决方案

为了解决这个问题，可以采取以下措施：

4.1 顺序消息

消息顺序性保证：虽然Kafka不保证全局消息顺序，但可以通过合理的分区策略和消息键来确保同一账单的消息被发送到同一个分区，从而在一定程度上保证消息的顺序性。

比如RocketMQ支持顺序消息。但是需要注意这是局部有序，非全局后续。具体实现过程：

1.发送mq消息时，通过selector将同一个业务主键的消息，发送到同一队列中

2.消费方使用MessageListenerOrderly消费局部有序的消息

该方案需要发送方和消费方同步改造。

4.2 前置检测

•在消费者处理消息之前，进行前置条件检查。例如，可以查询一个消息辅助表，确保上一个消息已经被成功消费或存入死信队列中。这种检查可以确保消息按照正确的顺序被处理。

•另一种方法是，在消息中添加序列号或时间戳，并在消费者端进行验证。如果当前消息的序列号或时间戳不符合预期顺序，则暂停处理并等待正确的消息到达。

4.3 状态机

在消息处理系统中，状态机可以用来定义和处理消息的顺序。每个状态代表系统当前所处的特定条件或阶段，而状态之间的转换则是由接收到的消息触发的。当系统接收到一个消息时，它会检查当前的状态和消息类型，然后决定是否要转移到另一个状态并执行相应的动作。

对于消息乱序问题，状态机可以通过以下方式解决：

1.定义状态转换规则：首先，需要定义一套明确的状态转换规则。这些规则应该基于业务逻辑来确定，以确保消息按照正确的顺序被处理。例如，如果系统要求先处理事件A再处理事件B，那么状态机就应该在接收到事件A后转移到能够处理事件B的状态。

2.状态检查与消息缓存：当系统接收到一个消息时，它会检查当前的状态是否允许处理该消息。如果当前状态不允许处理该消息（即消息的顺序不正确），则可以将该消息缓存起来，等待状态机转移到正确的状态后再进行处理。

3.状态转移与消息处理：一旦状态机转移到正确的状态，它就可以处理缓存中的消息。这可以确保消息按照正确的顺序被处理，即使它们最初是以乱序到达的。

4.4 监控与报警

建立系统的监控和报警机制，及时发现并处理消息错乱等异常情况。

通过采取以上措施，可以大大降低账单还款系统中消息错乱导致的问题，提高系统的稳定性和用户体验。

5. 小结

MQ消息乱序是分布式系统的常见难题，影响系统稳定性和业务一致性。本文深入解析问题根源，探讨了顺序消息、前置检查、状态机等实战解决方案，为实际开发中的问题解决提供有力参考。

文章中难免会有不足之处，希望读者能给予宝贵的意见和建议。谢谢！

审核编辑 黄宇