前言

近日偶然聊起消息队列，发现知识模糊又破碎，遂广泛查询资料，做了这么一篇非常浅显的总结，聊以充作入门参考资料吧。

下面几个问题，如果不能回答地很好，可以试着在文中找寻一下答案。（答案整理汇总在文末，个人理解，仅供参考）

• 消息队列在项目中解决了哪些问题？
• RocketMQ如何保证消息不丢失？
• RocketMQ如何解决消息重复问题？
• RocketMQ如何保证消息的有序性？

1 消息队列的一些概念

到底什么是消息队列？一个简单的概括就是：消息传输过程中使用队列来存储消息的组件，在程序开发中通常代指消息中间件，具有代表性的产品包括RocketMQ、Kafka、RabbitMQ。

1.1 消息队列基本角色

既然消息队列是存储消息的，那么这些消息一定有来源、有去处的。消息的来源我们称之为消息生产者，消息的去处称为之消息消费者。

1.2 消息队列应用场景

消息队列是消息存储的容器，但它存在的意义可远不止如此，随着互联网业务扩张、技术架构日益庞大和复杂化，消息队列承担着以下三个重任：异步、解耦、削峰。

异步处理

随着业务场景的丰富，一些服务的调用链路越来越长，以电商项目为例，开始时可能只有用户-订单处理-库存处理这么几步，后来发送短信、营销活动、积分活动等服务慢慢增加，如果仍然是同步处理机制，用户等待响应的时间会大大延长，体验非常糟糕。这种情况下，使用消息队列进行异步处理就非常必要了。

将一些用户不关心或者不是强烈关注的任务，例如消息通知、积分计算、优惠券扣减、数据分析等等从订单主流程中拆分为下游服务，而原来的订单流程只保留主要步骤，处理完成后直接将消息投递到消息队列中，下游服务监听到消息后进行并行处理。

下面这个示意图非常直观地展示了通过消息队列进行异步处理的优势：减少客户端请求等待时间，让下游服务异步并发处理，提升系统使用使用体验和整体性能。

服务解耦

上面说到，业务场景越来越丰富，订单服务的下游服务不断扩充，为了适配这些服务，订单服务可能要做修改，任何一个下游服务的变更都可能影响到订单服务，这对订单服务来说是不可接受的，频繁的修改不但是代码维护的噩梦，也对系统稳定性形成了挑战。

因此，选用消息队列来对服务进行解耦就是理所当然的了，订单服务只要订单相关消息投递到消息队列中，下游服务按照自身需要订阅即可，这样就将订单服务解放出来了。

流量削峰

在系统中，某些后端服务业务逻辑比较重，处理时间长，容易成为系统的瓶颈，如果某一时刻爆发式的流量打过来，系统可能就顶不住而导致全面性的服务不可用。这个时候，就需要一个引入一个“蓄水池”来做缓冲，消息队列本身就有存储消息的功能，用在这个场景下再合适不过了。

1.3 队列模型和发布/订阅模型

消息队列有两种模型：队列模型和发布/订阅模型。RabbitMQ 采用队列模型，RocketMQ和Kafka 采用发布/订阅模型。

队列模型

生产者往某个队列里面发送消息，一个队列可以存储多个生产者的消息，一个队列也可以有多个消费者， 但是消费者之间是竞争关系，即每条消息只能被一个消费者消费。队列模型可以通过消息全量存储至多个队列来解决一条消息被多个消费者消费问题，但代价是数据冗余。

发布/订阅模型

队列模式无法满足一条消息被多个消费者消费的需求，因此发布/订阅模式应运而生。该模型是将消息发往一个Topic即主题中，所有订阅了这个 Topic 的订阅者都能消费这条消息。

发布/订阅模型，我们可以把它理解为群聊，我发一条消息，加入了这个群聊的人都能收到这条消息，而队列模型就是一对一聊天，我发给你的消息，只能在你的聊天窗口弹出，别人是无法接收到的。

2 RocketMQ静态模型

RocketMQ是阿里巴巴旗下一款开源的分布式MQ框架，Java编程语言实现，有非常好完整生态系统，支持事务消息、顺序消息、批量消息、定时消息、消息回溯等特性。

2.1 消息的领域模型

消息的逻辑划分：Topic、Tags、Kyes

消息队列顾名思义，消息是其中的关键概念之一，消息是 RocketMQ 中的最小数据传输单元。生产者将业务数据的负载和拓展属性包装成消息发送到 RocketMQ 服务端，服务端按照相关语义将消息投递到消费端进行消费。

消息需要在逻辑上进行划分，其中Topic（主题）是必选的逻辑划分元素，而Tags、Keys是非必选，例如下面的代码示例：

Message msg = new Message( topic: "TopicTest", tags: "TagA", keys: "ORDER13214282571806",("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
);

Topic

主题是 RocketMQ 中消息传输和存储的顶层容器，用于标识同一类业务逻辑的消息。主题的作用主要如下：

• 定义数据的分类隔离： 在 RocketMQ 的方案设计中，建议将不同业务类型的数据拆分到不同的主题中管理，通过主题实现存储的隔离性和订阅隔离性。
• 定义数据的身份和权限： RocketMQ 的消息本身是匿名无身份的，同一分类的消息使用相同的主题来做身份识别和权限管理。

主题是一个逻辑概念，并不是实际的消息容器。主题内部由多个队列组成，消息的存储和水平扩展能力最终是由队列实现的；并且针对主题的所有约束和属性设置，最终也是通过主题内部的队列来实现。这一点在上面的领域模型图中有非常直观的体现。

RocketMQ 虽然提供了自动创建主题的功能，但是建议仅在测试环境使用，生产环境请勿打开，避免产生大量垃圾主题，无法管理和回收并浪费系统资源。

Tag

Topic 与 Tag 都是业务上用来归类的标识，区别在于 Topic 是一级分类，而 Tag 可以理解为是二级分类。使用 Tag 可以实现对 Topic 中的消息进行过滤。

Topic 和 Tag 的关系如下图所示。

什么时候该用 Topic，什么时候该用 Tag？

• 消息类型是否一致：如普通消息、事务消息、定时（延时）消息、顺序消息，不同的消息类型使用不同的 Topic，无法通过 Tag 进行区分。

• 业务是否相关联：没有直接关联的消息，如淘宝交易消息，京东物流消息使用不同的 Topic 进行区分；而同样是天猫交易消息，电器类订单、女装类订单、化妆品类订单的消息可以用 Tag 进行区分。

• 消息优先级是否一致：如同样是物流消息，盒马必须小时内送达，天猫超市 24 小时内送达，淘宝物流则相对会慢一些，不同优先级的消息用不同的 Topic 进行区分。

• 消息量级是否相当：有些业务消息虽然量小但是实时性要求高，如果跟某些万亿量级的消息使用同一个 Topic，则有可能会因为过长的等待时间而“饿死”，此时需要将不同量级的消息进行拆分，使用不同的 Topic。

Keys

RocketMQ 每个消息可以在业务层面的设置唯一标识码 keys 字段，方便将来定位消息丢失问题。 Broker 端会为每个消息创建索引（哈希索引），应用可以通过 topic、key 来查询这条消息内容，以及消息被谁消费。由于是哈希索引，请务必保证 key 尽可能唯一，这样可以避免潜在的哈希冲突。

RocketMQ 部署安装包默认开启了 autoCreateTopicEnable 配置，会自动为发送的消息创建 Topic，但该特性仅推荐在初期测试时使用。

生产环境强烈建议管理所有主题的生命周期，关闭自动创建参数，以避免生产集群出现大量无效主题，无法管理和回收，造成集群注册压力增大，影响生产集群的稳定性。

消息的物理划分：队列

与Topic、Tags、Kyes这些逻辑概念相对应，队列是 RocketMQ 中消息存储和传输的实际容器，也是 RocketMQ 消息的最小存储单元。 一个 Topic 可能有多个队列，并且可能分布在不同的 Broker 上，如下图所示：

总结如下：

1. 一个 Topic 有若干个队列（相当于分区），通常消息的生产快于消息的消费，在集群消费模式下（详见后文）设置多个队列可以提高消息消费的并行度，加快整体消费速度。
2. 一个Topic可以分布在不同的Broker上（称为 Topic 的分片），是为了突破单点的资源（CPU、带宽、内存等）限制，从而实现水平扩展。

最终的一个示意图如下：

2.2 RocketMQ部署模型

RocketMQ主要有四大核心组成部分：NameServer、Broker、Producer以及Consumer，具体如下图所示：

名字服务器 NameServer

NameServer是一个简单的 Topic 路由注册中心，支持 Topic、Broker 的动态注册与发现。

主要包括两个功能：

• Broker管理，NameServer接受Broker集群的注册信息并且保存下来作为路由信息的基本数据。然后提供心跳检测机制，检查Broker是否还存活；
• 路由信息管理，每个NameServer将保存关于 Broker 集群的整个路由信息和用于客户端查询的队列信息。Producer和Consumer通过NameServer就可以知道整个Broker集群的路由信息，从而进行消息的投递和消费。

NameServer 通常会有多个实例部署，各实例间相互不进行信息通讯。Broker 是向每一台 NameServer 注册自己的路由信息，所以每一个 NameServer 实例上面都保存一份完整的路由信息。当某个 NameServer 因某种原因下线了，客户端仍然可以向其它 NameServer 获取路由信息。

代理服务器 Broker

Broker 主要负责消息的存储、投递和查询以及服务高可用保证。NameServer几乎无状态节点，因此可集群部署，节点之间无任何信息同步。Broker 部署相对复杂。

Broker 是 RocketMQ 的核心模块，负责接收并存储消息，同时提供 Push/Pull 接口来将消息发送给 Consumer。Consumer 可选择从Master或者Slave读取数据。多个主/从组成Broker集群，集群内的Master节点之间不做数据交互。Broker同时提供消息查询的功能，可以通过MessageID和MessageKey来查询消息。Borker会将自己的Topic配置信息实时同步到NameServer。

在 Master-Slave 架构中，Broker 分为 Master 与 Slave。一个Master可以对应多个Slave，但是一个Slave只能对应一个Master。Master 与 Slave 的对应关系通过指定相同的BrokerName，不同的BrokerId 来定义，BrokerId为0表示Master，非0表示Slave。Master也可以部署多个。

生产者 Producer

消息生产者，会先和NameServer集群中的随机一台建立长连接，获取当前待发送的 Topic 存储在哪台 Broker Master上，然后再与其建立长连接。Producer位于用户进程内，多个发送同一类消息的生产者称之为一个生产者组，即Producer Group。

消费者 Consumer

消息消费者，同Producer一样，Consumer先通过NameServer获取所有broker的路由信息后，向Broker发送Pull请求来获取消息数据。消费同一类消息的多个 Consumer 实例组成一个消费者组，即Consumer Group。

部署模型小结

• 每个 Broker 与 NameServer 集群中的所有节点建立长连接，定时注册 Topic 信息到所有NameServer。

• Producer 与 NameServer 集群中的其中一个节点建立长连接，定期从 NameServer 获取Topic路由信息，并向提供Topic 服务的 Master 建立长连接，且定时向 Master 发送心跳。Producer 完全无状态。

• Consumer 与 NameServer 集群中的其中一个节点建立长连接，定期从 NameServer 获取 Topic路由信息，并向提供 Topic 服务的 Master、Slave 建立长连接，且定时向 Master、Slave发送心跳。Consumer既可以从 Master 订阅消息，也可以从Slave订阅消息。

RocketMQ集群工作流程：

1. 启动NameServer：启动NameServer。NameServer启动后监听端口，等待Broker、Producer、Consumer连接，相当于一个路由控制中心。
2. 启动 Broker：启动 Broker。与所有 NameServer 保持长连接，定时发送心跳包。心跳包中包含当前 Broker 信息以及存储所有 Topic 信息。注册成功后，NameServer 集群中就有 Topic跟Broker 的映射关系。
3. 创建 Topic：创建 Topic 时需要指定该 Topic 要存储在哪些 Broker 上，也可以在发送消息时自动创建Topic，但生产环境不建议这么做。
4. 生产者发送消息：生产者启动时先跟 NameServer 集群中的其中一台建立长连接，并从 NameServer 中获取当前发送的 Topic存在于哪些 Broker 上，轮询从队列列表中选择一个队列，然后与队列所在的 Broker建立长连接从而向 Broker发消息。
5. 消费者接受消息：消费者跟其中一台NameServer建立长连接，获取当前订阅Topic存在哪些Broker上，然后直接跟Broker建立连接通道，然后开始消费消息。

3 RocketMQ消息的流转过程

3.1 消息的生产

消息发送的负载均衡

在前文的模型中，我们知悉：

1. 一个 Topic 的消息可能分布在不同的 Broker 的多个队列中。
2. 生产者发送消息时只指定了 NameServer 的地址，并没有具体的 Broker信息。

于是，我们要面对这个问题：生产者如何知悉自己的消息要发送到哪个队列，这个队列所属的Broker服务器地址信息又如何获得？

这里，RocketMQ 的解决方案是两步解决：

首先获取Topic对应的所有队列消息

NameServer 中会维护一个路由表，包含所有Broker的信息（含Broker机器的IP和端口），以及Broker和Topic、队列的对应关系。

当生产者启动的时候，会从NameServer中拉取到路由表，缓存到本地，同时会开启一个定时任务，默认是每隔30s从NameServer中重新拉取路由信息，更新本地缓存。

其次，面对多个队列的选择问题，RocketMQ提供了两种消息队列的选择算法。

• 轮询算法
• 最小投递延迟算法

轮询算法 就是一个队列一个队列发送消息，这些就能保证消息能够均匀分布在不同的队列底下，这也是RocketMQ默认的队列选择算法。至于最小投递延迟算法，就是选择投递延迟时间最小的队列，这种算法会导致消费发布不均匀的问题。

消息的同步、异步、单向传输

RocketMQ可用于以三种方式发送消息：同步、异步和单向传输。前两种消息类型是可靠的，因为无论它们是否成功发送都有响应。

3.2 消息的存储

消息从消费者发出后，到达某个 Broker 中，先要解决消息存储的问题，RocketMQ 设计了三种文件来存储消息：CommitLog、ConsumeQueue、IndexFile。

消息在磁盘中如何存储？

1. CommitLog

   这个文件是消息主体以及元数据的存储主体，存储生产端写入的消息主体内容，包括消息本身的内容数据、消息的Topic、消息所在队列的id、消息生产者的ip和端口等。CommitLog在物理磁盘文件上被分为多个磁盘文件，每个文件默认的固定大小是1G
   

2. ConsumeQueue

   当消息存储到 CommitLog 中后，ConsumeQueue 中也会同步插入一条数据，记录：消息在CommitLog的起始位置、消息在CommitLog存储的长度、消息tag的hashCode。

   消费者拉取消息的时候，服务端会先去ConsumeQueue 中查找消费者需要的消息在CommitLog中的位置，然后再去CommitLog文件拉取真正的消息内容。

   所以，从这可以看出，ConsumeQueue其实就相当于是一个索引文件，方便我们快速查找在CommitLog中的消息。
   

3. IndexFile

   索引文件提供了一种可以通过 key 或时间区间来查询消息的方法。

总结下来就是，消息到了先存储到 Commitlog，然后会有一个 ReputMessageService 线程接近实时地将消息转发给 ConsumeQueue 文件与 IndexFile，也就是说是异步生成的。

消息存储的刷盘机制

RocketMQ 在将消息写到CommitLog文件中时并不是直接就写到文件中，而是先写到PageCache，也就是前面说的内核缓存区，所以RocketMQ 提供了两种刷盘机制，来将内核缓存区的数据刷到磁盘。

• 异步刷盘：Broker将消息写到PageCache的时候，就直接返回给生产者说消息存储成功了，然后通过另一个后台线程来将消息刷到磁盘，这个后台线程是在RokcetMQ启动的时候就会开启。异步刷盘方式也是RocketMQ默认的刷盘方式。

• 同步刷盘就是指Broker将消息写到PageCache的时候，会等待异步线程将消息成功刷到磁盘之后再返回给生产者说消息存储成功。

同步刷盘相对于异步刷盘来说消息的可靠性更高，因为异步刷盘可能出现消息并没有成功刷到磁盘时，机器就宕机的情况，此时消息就丢了；但是同步刷盘需要等待消息刷到磁盘，那么相比异步刷盘吞吐量会降低。所以同步刷盘适合那种对数据可靠性要求高的场景。

如果你需要使用同步刷盘机制，只需要在配置文件指定一下刷盘机制即可。

最后以一张图作为消息流转全过程的总结：

3.3 消息的消费

当消费在 Broker 存储完毕，消费者要获取消息进行消费了。在这个环节，将面临以下几个问题：

1. 消息推送还是拉取：消费者如何获取消息，是主动到 Broker 拉取消息还是等待 Broker 推送消息？
2. 消息消费模式：一个消息是被所有消费者消费，还是只会被一个消费者消费？
3. 消费者的负载均衡问题：如果消息只被一个消费者消费，多个消费者如何分配消息？
4. 消费位点：如果一个队列有多个消费者订阅，怎么管理这些不同消费者的消费进度呢？
5. 死信队列：如果消息经过多次消费重试依然无法正常处理，怎么办呢？

消息的推送与拉取

推拉模式各有优缺点：

• 推模式：优势是消息实时性高，Broker收到消息后立即推送给消费者，缺点是推模式难以兼顾消费者实际消费情况，如果消息生产速率高于消费速率，推模式会导致消费者压力过大。
• 拉模式：这种模式消费者能主动控制消息拉取频率，按需消费，缺点是消息拉取可能有延迟，另外会存在大量拉取请求在做无用功。

权衡利弊，RocketMQ 采用的是”长轮询“方式实现的拉模式，在利用拉模式优势的同时尽量规避其缺点。

广播消费和共享消费

在 RocketMQ 领域模型中，同一条消息支持被多个消费者分组订阅，同时，对于每个消费者分组可以初始化多个消费者。您可以根据消费者分组和消费者的不同组合，实现以下两种不同的消费效果：

• 消费组间广播消费 ：如上图所示，每个消费者分组只初始化唯一一个消费者，每个消费者可消费到消费者分组内所有的消息，各消费者分组都订阅相同的消息，以此实现单客户端级别的广播一对多推送效果。该方式一般可用于网关推送、配置推送等场景。

• 消费组内共享消费 ：如上图所示，每个消费者分组下初始化了多个消费者，这些消费者共同分担消费者分组内的所有消息，实现消费者分组内流量的水平拆分和均衡负载。该方式一般可用于微服务解耦场景。

消费者负载均衡

消费组间广播消费场景下，每个消费者分组内只有一个消费者，因此不涉及消费者的负载均衡。

消费组内共享消费场景下，消费者分组内多个消费者共同分担消息，消息按照哪种逻辑分配给哪个消费者，就是由消费者负载均衡策略所决定的。

根据消费者类型的不同，消费者负载均衡策略分为以下两种模式：

• 消息粒度负载均衡：PushConsumer和SimpleConsumer默认负载策略

• 队列粒度负载均衡：PullConsumer默认负载策略

消息粒度负载均衡

消息粒度负载均衡策略中，同一消费者分组内的多个消费者将按照消息粒度平均分摊主题中的所有消息，即同一个队列中的消息，可被平均分配给多个消费者共同消费。

如上图所示，消费者分组Group A中有三个消费者A1、A2和A3，这三个消费者将共同消费主题中同一队列Queue1中的多条消息。 注意 消息粒度负载均衡策略保证同一个队列的消息可以被多个消费者共同处理，但是该策略使用的消息分配算法结果是随机的，并不能指定消息被哪一个特定的消费者处理。

消息粒度的负载均衡机制，是基于内部的单条消息确认语义实现的。消费者获取某条消息后，服务端会将该消息加锁，保证这条消息对其他消费者不可见，直到该消息消费成功或消费超时。因此，即使多个消费者同时消费同一队列的消息，服务端也可保证消息不会被多个消费者重复消费。

队列粒度负载均衡

队列粒度负载均衡策略中，同一消费者分组内的多个消费者将按照队列粒度消费消息，即每个队列仅被一个消费者消费。

如上图所示，主题中的三个队列Queue1、Queue2、Queue3被分配给消费者分组中的两个消费者，每个队列只能分配给一个消费者消费，该示例中由于队列数大于消费者数，因此，消费者A2被分配了两个队列。若队列数小于消费者数量，可能会出现部分消费者无绑定队列的情况。

队列粒度的负载均衡，基于队列数量、消费者数量等运行数据进行统一的算法分配，将每个队列绑定到特定的消费者，然后每个消费者按照取消息>提交消费位点>持久化消费位点的消费语义处理消息，取消息过程不提交消费状态，因此，为了避免消息被多个消费者重复消费，每个队列仅支持被一个消费者消费。

相对于消息粒度负载均衡策略，队列粒度负载均衡策略分配粒度较大，不够灵活。但该策略在流式处理场景下有天然优势，能够保证同一队列的消息被相同的消费者处理，对于批量处理、聚合处理更友好。

消费进度管理

RocketMQ 领域模型为发布订阅模式，每个主题的队列都可以被多个消费者分组订阅。若某条消息被某个消费者消费后直接被删除，则其他订阅了该主题的消费者将无法消费该消息。

因此，Apache RocketMQ 通过消费位点管理消息的消费进度。每条消息被某个消费者消费完成后不会立即在队列中删除，Apache RocketMQ 会基于每个消费者分组维护一份消费记录，该记录指定消费者分组消费某一个队列时，消费过的最新一条消息的位点，即消费位点。

当消费者客户端离线，又再次重新上线时，会严格按照服务端保存的消费进度继续处理消息。如果服务端保存的历史位点信息已过期被删除，此时消费位点向前移动至服务端存储的最小位点。

死信队列

对于消费失败且重试后依然失败的消息， RocketMQ 不会立丢弃，而是将消息转发至指定的队列中，即死信队列，这些消息即为死信消息。当消费失败的原因排查并解决后，您可以重发这些死信消息，让消费者重新消费；若您暂时无法处理这些死信消息，为避免到期后死信消息被删除，您也可以先将死信消息导出进行保存。

死信消息具有以下特性：

• 不会再被消费者正常消费。
• 有效期与正常消息相同，默认为3天，3天后会被自动删除。因此，请在死信消息产生后的3天内及时处理。

4 RocketMQ中典型问题处理

4.1 如何保证消息不丢失

一条消息的全部流转过程，无非是三个阶段：生产消息、存储消息和消费消息，如何防止消息丢失就需要从这三个阶段入手。

• 生产消息阶段：生产者发送消息到 Broker ，需妥善处理 Broker 响应，无论同步发送还是异步发送都需要做好异常处理，如果 Broker 没有在约定时间内返回接收成功的消息，就需要生产者进行重试，重试要有记录。

• 存储消息阶段：存储消息阶段需要考虑刷盘可靠性和高可用。刷盘机制有同步刷盘和异步刷盘，同步刷盘是牺牲效率保障可靠性，对于重要性高的消息要启用同步刷盘机制。高可用主要依赖 RocketMQ 本身的主从机制。

• 消费消息阶段：这个阶段可能的风险在于有开发者收到 Broker 的消息后立即返回消费完成的响应，而后续这条消息并不一定能成功处理，正确的做法是我们在消息业务逻辑处理完成之后再给Broker响应，这样消息就不会在消费阶段丢失。

4.2 如何解决消息重复问题

首先，实际生产中，消息重复的问题无法彻底规避。

如上图，在消息发送阶段和消息消费阶段，因为要保证可靠性，都有重试机制，这恰恰是重复消息产生的温床。

比如生产者发送消息后，Broker已经收到并存储了消息，但由于网络原因响应没有返回或返回超时，生产者进行重试，于是重复消息发送到了 Broker。

再比如，消费者将消息消费了，业务逻辑已经走完了，此时需要更新Consumer offset了，然后这个消费者挂了，另一个消费者顶上，此时Consumer offset还没更新，于是又拿到刚才那条消息，业务又被执行了一遍，于是消息又重复了。

既然我们不能规避重复消息的产生，那么我们只能在业务上消除重复消息所带来的影响，而关键就是幂等设计。关于幂等，常见手段有分布式锁、乐观锁、数据库唯一键约束（insert into update on duplicate key）等，这里不做展开。

4.3 如何保证消息的有序性

消息的有序性分为全局有序和部分有序。前者要求只能有一个生产者发送消息，且 Topic 只能有一个队列，鉴于全局有序很少有应用场景且实现代价高，我们一般不考虑，重点是保障局部有序性。

局部有序性指的是，强相关的一组消息内部保持有序性，而组与组之间不保证顺序。举个例子，用户A的订单、支付、库存消息一定要严格按照顺序处理，不能出现库存消息跑到了订单消息之前，但用户A和用户B的消息顺序无需特殊处理，哪怕用户A的订单消息早于用户B的订单消息进入系统，但系统率先处理了B的订单，也是可以接受的。

如何保证局部有序性呢？这也需要生产者和消费者两端共同配合。一句话概括就是：一个生产者只向一个特定的队列发送消息，一个队列只被一个特定的消费者单线程消费消息。

首先是生产端，把某一组消息通过特定的策略发往固定的队列中。实际业务中一个思路是将用户按id分桶，每组用户绑定一个队列，这样一个用户的所有消息必然在一个队列中。

RocketMQ提供了自定义队列选择的接口 MessageQueueSelector：

for (int i = 0; i < 100; i++) {int orderId = i % 10;Message msg = new Message("TopicTest", tags[i % tags.length], "KEY" + i,("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));SendResult sendResult = producer.send(msg, new MessageQueueSelector() {@Overridepublic MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {Integer id = (Integer) arg;int index = id % mqs.size();return mqs.get(index);}}, orderId);System.out.printf("%s%n", sendResult);
}

其次是消费端，关键点是一个队列的消息要被一个固定的消费者单线程处理，注意以下两点：

• 要使用队列粒度负载均衡，即一个队列的消息只被一个固定的消费者消费，而消息粒度负载均衡策略会使一个队列的消息被多个消费者消费的情况
• 要保证消费者单线程消费，在并发消费中，可能会有多个线程同时消费一个队列的消息，因此即使发送端通过发送顺序消息保证消息在同一个队列中按照FIFO的顺序，也无法保证消息实际被顺序消费。

RocketMQ 提供了顺序消费的方式， 顺序消费设置与并发消费API层面只有一处不同，在注册消费回调接口时传入MessageListenerOrderly接口的实现：

consumer.registerMessageListener(new MessageListenerOrderly() {AtomicLong consumeTimes = new AtomicLong(0);@Overridepublic ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);this.consumeTimes.incrementAndGet();if ((this.consumeTimes.get() % 2) == 0) {return ConsumeOrderlyStatus.SUCCESS;} else if ((this.consumeTimes.get() % 5) == 0) {context.setSuspendCurrentQueueTimeMillis(3000);return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;}return ConsumeOrderlyStatus.SUCCESS;}
});

RocketMQ 的丰富内容让人望而生畏，只能总结这么一点了，至于延时消息、事务消息等，复杂度太高，就先不涉及了，留待有缘人补充。