面试官问：如何保证 MQ消息是有序的？

Posted 2021-09-18 Hollis Chuang

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了面试官问：如何保证 MQ消息是有序的？相关的知识，希望对你有一定的参考价值。

为了系统间解耦，我们通常会引入MQ框架，大家各司其职共同完成上下游的业务流程。

大致过程：

生产端，创建一条消息，通过网络发送到MQ Server
MQ将消息存储在topic 的一个分区里
消费端，从分区中拉取消息，消费处理

但现实往往不一样！MQ 架构设计要满足高并发、高性能、高可用等指标

单分区，达不到我们的吞吐量要求，我们考虑采用多分区架构设计，正所谓 ”三个臭皮匠赛过一个诸葛亮“，多分区可以有效分摊全局压力，提升整体系统性能。

两台 MQ机器，组成一个集群，原先一个分区存储6条消息，现在分摊到两个分区，每个分区各存储3条消息，性能比上面那个提升一倍。

貌似可以满足我们的需求，但任何事情都有两面性！

我们看看下面业务场景：

一个用户在电商网站上下订单到交易完成，中间会经历一系列动作，订单的状态也会随之变化，一个订单会产生多条MQ消息，下单、付款、发货、买家确认收货，消费端需要严格按照业务状态机的顺序处理，否则，就会出现业务问题。

我们发现，消息带上了状态，不再是一个个独立的个体，有了上下文依赖关系！

对于这个问题，突然想到HTTP协议，其本身也是无状态的，也就是说前后两次请求没有关联，但有些业务功能有登录要求，那怎么解决？

引入Cookie机制，每次请求客户端额外传输一些数据，来达到上下文关联。

回到MQ的消息顺序问题，我们要如何解决？

答案：各退一步，保证局部有序。

比如上面的电商例子，只要保证一个订单的多条状态消息在同一个分区，便可以满足业务需求，这个方案可以覆盖大部分的业务场景。

这里面只需要有一个路由策略组件，由它决定消息该放到哪个分区中！

考虑到市面MQ开源框架很多，常见的如：Kafka、Pulsar、RabbitMQ、RocketMQ 等，API方法略有区别，但设计思路是相通的。

接下来，我们以 RocketMQ 为例：

生产端提供了一个接口 MessageQueueSelector

public interface MessageQueueSelector {
   MessageQueue select(final List<MessageQueue> mqs, final Message msg, final Object arg);
}

接口内定义一个select方法，具体参数含义：

mqs：该Topic下所有的队列分片
msg：待发送的消息
arg：发送消息时传递的参数

关于MessageQueueSelector接口，RocketMQ 框架提供了三个默认实现类：

1、SelectMessageQueueByHash：

arg参数的hashcode的绝对值，然后对mqs.size()取余，得到目标队列在mqs的下标

2、SelectMessageQueueByRandom：

对mqs.size()值取随机数作为目标队列在mqs的下标

3、SelectMessageQueueByMachineRoom

返回null

特别注意：

虽然保证了单个分片的消息有序，但每个分片的消费者只能是单线程处理，因为多线程无法控制消费顺序。这个可能会损失一些性能。

这里又引出另一个问题，如何保证一个队列只能有一个消费端呢？

1、

org.apache.rocketmq.client.impl.consumer.RebalanceImpl#updateProcessQueueTableInRebalance

遍历一个topic下所有的MessageQueue
isOrder && !this.lock(mq) 尝试对它加锁，确保一个MessageQueue只能被一个消费者处理

2、将PullRequest对象放入PullMessageService的pullRequestQueue队列中

public void dispatchPullRequest(List<PullRequest> pullRequestList) {
    for (PullRequest pullRequest : pullRequestList) {
        this.defaultMQPushConsumerImpl.executePullRequestImmediately(pullRequest);
        log.info("doRebalance, {}, add a new pull request {}", consumerGroup, pullRequest);
    }
}

3、org.apache.rocketmq.client.impl.consumer.PullMessageService#run