实战!如何保证 MQ消息是有序的?

实战!如何保证 MQ消息是有序的?

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

QQ截图20211216085755

大致过程:

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

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

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

QQ截图20211216090030

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

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

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

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

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

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

”

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

QQ截图20211216090253

答案:各退一步,保证局部有序。

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

”

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

考虑到市面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

640

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

2、将PullRequest对象放入PullMessageServicepullRequestQueue队列中

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

640 (1)

  • PullMessageService 是一个Runnable线程任务
  • 无限循环,从队列中拉取、处理消息

另一个问题,如何保证一个队列,只有一个线程在处理消息呢?

1、DefaultMQPushConsumerImpl#pullMessage

640 (2)

  • ConsumeMessageService 中有两个实现类,因为我们有消费顺序要求,会选择ConsumeMessageOrderlyService来处理业务

2、ConsumeMessageOrderlyService.ConsumeRequest

640 (3)

  • ConcurrentMap中获取messageQueue对应的锁对象
  • 通过 synchronized 关键字,线程来抢占锁,互斥关系,从而保证了一个MessageQueue只能有一个线程并发处理

继续往下看,如果扩容了怎么办?

原来有6个分区,order_id_1的消息在MessageQueue6 中,此时扩容一倍,现在12个分区,order_id_1订单后面产生的消息可能路由到了MessageQueue8 中,同一个订单的消息分布在两个分区中,无法保证顺序。

我们能做的是,先将存量消息处理完,再扩容。如果是在线业务,可以搞个临时topic,先将消息暂时堆积,待扩容后,按新的路由规则重新发送。

顺序消息,如果某条失败了怎么办?会不会一直阻塞?

1、如果失败,不会提交消费位移,系统会自动重试(有重试上限),此时会阻塞后面的消息消费,直到这条消息处理完

2、如果这个消息达到重试上限,依然失败,会进入死信队列,可以继续处理后面的消息


已有 0 条评论

    欢迎您,新朋友,感谢参与互动!