mq如何保证消息顺序「tcp怎么保证有序传输」
今天给大家普及一下mq如何保证消息顺序「tcp怎么保证有序传输」相关知识,最近很多在问mq如何保证消息顺序「tcp怎么保证有序传输」,希望能帮助到您。
原文链接:https://mp.weixin.qq.com/s/V2PgGtHV5p7igoWC_qwHug
原作者:微观技术
为了系统间解耦,我们通常会引入MQ框架,大家各司其职共同完成上下游的业务流程。
大致过程:
生产端,创建一条消息,通过网络发送到MQ ServerMQ将 消息存储在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下所有的MessageQueueisOrder && !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
PullMessageService 是一个Runnable线程任务无限循环,从队列中拉取、处理消息另一个问题,如何保证一个队列,只有一个线程在处理消息呢?
1、 DefaultMQPushConsumerImpl#pullMessage
ConsumeMessageService 中有两个实现类,因为我们有消费顺序要求,会选择ConsumeMessageOrderlyService来处理业务2、 ConsumeMessageOrderlyService.ConsumeRequest
从ConcurrentMap中获取messageQueue对应的锁对象通过 synchronized 关键字,线程来抢占锁,互斥关系,从而保证了一个MessageQueue只能有一个线程并发处理继续往下看,如果扩容了怎么办?
原来有6个分区,order_id_1的消息在MessageQueue6 中,此时扩容一倍,现在12个分区,order_id_1订单后面产生的消息可能路由到了MessageQueue8 中,同一个订单的消息分布在两个分区中,无法保证顺序。
我们能做的是,先将存量消息处理完,再扩容。如果是在线业务,可以搞个临时topic,先将消息暂时堆积,待扩容后,按新的路由规则重新发送。
顺序消息,如果某条失败了怎么办?会不会一直阻塞?
1、如果失败,不会提交消费位移,系统会自动重试(有重试上限),此时会阻塞后面的消息消费,直到这条消息处理完
2、如果这个消息达到重试上限,依然失败,会进入死信队列,可以继续处理后面的消息