在分布式系统和高并发应用场景中，消息队列（MQ）扮演着至关重要的角色。RocketMQ作为阿里巴巴开源的一款高性能、高吞吐量的消息中间件，广泛应用于微服务架构中，以实现消息的可靠异步通信。然而，在实际应用中，RocketMQ的消息顺序问题若处理不当，往往会引发线上事故。本文将详细记录一次由于RocketMQ消费非顺序消息引起的线上事故，并通过C#例子代码展示如何避免类似问题的发生。

一、背景介绍

公司最近引入了阿里云的RocketMQ企业铂金版本，由阿里云平台完全托管。RocketMQ能够在微服务、分布式系统和无服务应用程序之间，提供基于消息的可靠异步通信机制，帮助构建松耦合、可扩展、高可用分布式系统。然而，在系统的实际运行过程中，由于并发量大、消息处理逻辑复杂等因素，我们遇到了一次因消息消费非顺序导致的线上事故。

二、事故经过

事故发生在某个订单处理系统中，该系统依赖于RocketMQ进行订单消息的异步处理。订单处理流程包括订单创建、订单支付、订单发货等多个环节，每个环节都会产生相应的消息发送到RocketMQ。消费者从RocketMQ拉取消息，并按顺序处理这些消息，以确保订单处理的正确性。

然而，在系统运行一段时间后，我们发现部分订单的处理顺序出现了错乱，导致订单状态不一致，进而影响到了用户的购物体验。经过排查，我们发现问题出在RocketMQ的消息消费上。由于RocketMQ默认不支持严格的顺序消息，且我们在消费者端没有做好相应的处理，导致消息被并发消费，从而出现了顺序错乱的情况。

三、问题分析

RocketMQ默认不支持严格的顺序消息：RocketMQ的一个Topic可以有多个Queue，且消费是多线程进行的。即使Queue能够保证先进先出，但由于多线程CPU执行的随机性，我们无法保证消费的顺序。
消费者端处理逻辑不当：在消费者端，我们没有对需要保持顺序的消息进行特殊处理，而是简单地使用并发消费模式，导致消息被乱序处理。
系统并发量大：在系统并发使用量较大的情况下，RocketMQ的性能虽然能够满足需求，但消息的消费顺序却难以保证。

四、解决方案

针对上述问题，我们提出了以下解决方案：

确保消息顺序消费：

以下是一个C#示例代码，展示如何在生产者和消费者端使用MessageQueueSelector：

// 生产者代码
public class Producer
{
    public static void Main(string[] args)
    {
        // ... 初始化生产者 ...

        MessageQueueSelector selector = new MessageQueueSelector((mqs, msg, arg) =>
        {
            long orderId = (long)arg;
            int index = (int)(orderId % mqs.Count);
            return mqs[index];
        });

        // 发送消息，指定MessageQueueSelector和订单号作为参数
        producer.Send(message, selector, orderId);

        // ... 关闭生产者 ...
    }
}

// 消费者代码
public class Consumer
{
    public static void Main(string[] args)
    {
        // ... 初始化消费者，设置订阅的Topic和Tag ...

        consumer.Subscribe("orderTopic", "*");

        // 设置消息监听器，这里使用顺序消费模式
        consumer.RegisterMessageListener(new MessageListenerOrderly(ConsumeMessage));

        // 启动消费者
        consumer.Start();

        // ... 消息消费逻辑 ...

        static ConsumeOrderlyStatus ConsumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context)
        {
            // 按顺序处理消息
            foreach (var msg in msgs)
            {
                // 处理消息逻辑
            }

            return ConsumeOrderlyStatus.SUCCESS;
        }
    }
}

设置单个队列：在RocketMQ的dashboard中，将当前Topic的writeQueueNums和readQueueNums设置为1，这样当前Topic中只有一个队列，所有消息都顺序保存和消费。但这种方法会增加消息存储和消费的压力。
使用MessageQueueSelector：在生产者和消费者代码中，通过MessageQueueSelector接口选择消息队列，确保需要顺序消费的消息被发送到同一个队列中。具体实现可以通过取模运算（如订单号%队列数）来确定消息应发送到的队列。

增加幂等性处理：

由于网络问题或消费者重启等原因，可能会导致消息被重复消费。为了避免数据不一致的问题，我们需要在消费者端增加幂等性处理。具体实现可以通过维护一个已处理消息的记录表，在消费消息前查询该记录表，如果消息已处理则跳过，否则进行业务处理并记录消息ID。

优化系统架构：

对于并发量大的系统，除了确保消息顺序消费外，还需要优化系统架构，如引入更多的消费者实例、优化数据库性能、使用缓存等，以提高系统的整体处理能力和响应速度。

五、总结与反思

本次线上事故给我们敲响了警钟，提醒我们在使用RocketMQ等消息队列时，必须充分了解其特性和限制，并采取相应的措施来确保消息的顺序性和幂等性。同时，我们也需要不断优化系统架构和代码实现，以应对日益增长的并发量和业务需求。

在未来的工作中，我们将继续加强对RocketMQ等中间件的学习和研究，探索更多优化和解决问题的方案。同时，我们也将建立完善的监控和报警机制，及时发现并处理潜在的线上问题，确保系统的稳定运行和用户的良好体验。

ITPub博客

记一次RocketMQ消费非顺序消息引起的线上事故

最新文章