消息队列——引入的问题：重复消费&顺序消费&分布式事务

墨蓝 2023-09-28 16:04 80阅读 0赞

上篇文章[《消息队列入门》][Link 1]里面提到，**技术是把双刃剑**！使用消息队列是因为他带给我们很多好处：系统间解耦、异步执行、削峰限流，但是使用消息队列可能带来一些问题：消息队列的消息被重复消费、消息的顺序保证、异步执行导致的数据不一致；

本篇就来介绍一下这几个引入消费队列后可能带来的问题，以及常见的解决方案；

### 1. 消息重复消费问题 ###

消息重复消费是使用消息队列之后，必须考虑的一个问题，也是比较严重和常见的问题；

例如，对于订单场景，订单结果的通知一般都是异步方式，由订单中台接收第三方支付渠道的回执，然后封装成业务订单结果投递给业务方；目前我接触的投递方式有HTTP/DUBBO回调和消息队列MQ；

考虑到下游业务方的消费能力（消费速度），使用消息队列是比较好的选择；例如一个累计消费活动的页面，你下单完成了立刻去看一眼活动页面，有时候累计金额马上就加上去了，有时候却延迟有很久，为啥？这个速度取决于消息队列的消费速度，消费慢堵塞了就迟点看到呗；

一般来说，消息队列的接收模式往往是手动确认的，这意味着如果消费者的处理逻辑中抛出了异常，该消息会被重新投递；

此外，支付结果消息的下游监听服务一般是多个，当某个下游业务方出现短期不可用，当它恢复时可能希望某个时间区间消息重新投递，这时其他系统也会收到这些被消费过的消息，就需要考虑其他系统收到重复消息时，是否会产生脏数据？如同一条支付成功的消息投递了两次，会不会出现发两个奖品的情况；

事实上，消息重发是很常见的，服务的**网络抖动**，**开发人员代码Bug**等都可能处理失败，并要求消息重发的；这就需要让我们的接口具备一个特性——**幂等**；

> 幂等（idempotent、idempotence）是一个数学与计算机学概念，常见于抽象代数中；
> 
> 在编程中一个幂等操作的特点是其任意多次执行所产生的影响均与一次执行的影响相同；
> 
> 幂等函数，或幂等方法，是指可以使用相同参数重复执行，并能获得相同结果的函数；这些函数不会影响系统状态，也不用担心重复执行会对系统造成改变；
> 
> ——[《幂等 百度百科》][Link 2]

#### **如何实现幂等？** ####

一般幂等，需要分场景去考虑，看是强校验还是弱校验；比如跟金钱相关的场景那就很关键，就做强校验，别不是很重要的场景如消息频控（疲劳度控制），就可以做弱校验；

**（1）强校验**：比如你监听到用户支付成功的消息，然后执行业务订单更新、权益发放、下游消息本地持久化这一系列DB操作，可以将这些操作放在一个事务内；

每次消息过来都要拿着业务订单号这样的唯一标识去订单流水表查，看看有没有已经处理过这条流水，处理过就直接终止流程返回，不用走下面的流程了，没有就执行后面的后置处理逻辑；

**（2）弱校验**：这个简单，一些不重要的场景，比如短信发放消息的处理，可以把\{接收者ID+业务场景ID\}作为唯一标识作为Redis的key放到缓存里面，设置失效时间（做疲劳度控制），每次接收到执行短信发放消息时就去Redis判断，看当前是否还需要给用户发短信；

之所以是弱校验，是因为用KV就算Redis内数据清空了也没关系，最多是多消费一次多发一条短信，业务也无关痛痒；

### 2. 消息顺序消费问题 ###

关于顺序消费，开发过程中这样的场景不多，网上更多的都是介绍binlog的同步，好像更多的场景就没了；正常的业务场景就是按照步骤执行，天然的时序的，很难出现并行投递消息的场景；

举个例子，我们都知道数据量大的时候，数据同步压力还是很大的，有时候数据量大的表需要同步几个亿的数据给下游，这种情况可以将需要同步的数据怼到队列里面去，然后下游慢慢消费的；

这时就可能出现同时对库里的一条数据进行了增、改、删三个操作，但是你消息发过去消费的时候变成了改、删、增，这样数据就不对了；

#### **如何解决消息顺序消费？** ####

几种不同MQ实现顺序消息的方式可以参考这篇[《几种MQ顺序消息的实现方式》][MQ]；这里简单的说一下**RocketMQ**里面的一个简单实现；

RocketMQ的[官方文档][Link 3]里面给出了"消息有序性"的定义——

> 消息有序指的是一类消息消费时，能按照发送的顺序来消费；

例如：一个订单产生了三条消息分别是订单创建、订单付款、订单完成，消费时要按照这个顺序消费才能有意义，但是同时订单之间是可以并行消费的；RocketMQ可以严格的保证消息有序；

顺序消息分为**全局顺序消息**与**分区顺序消息**，全局顺序是指某个Topic下的所有消息都要保证顺序；部分顺序消息只要保证每一组消息被顺序消费即可。

*  全局顺序 对于指定的一个 Topic，所有消息按照严格的先入先出（FIFO）的顺序进行发布和消费；适用场景：性能要求不高，所有的消息严格按照 FIFO 原则进行消息发布和消费的场景；
 *  分区顺序 对于指定的一个 Topic，所有消息根据 sharding key 进行区块分区； 同一个分区内的消息按照严格的 FIFO 顺序进行发布和消费；Sharding key 是顺序消息中用来区分不同分区的关键字段，和普通消息的 Key 是完全不同的概念； 适用场景：性能要求高，以 sharding key 作为分区字段，在同一个区块中严格的按照 FIFO 原则进行消息发布和消费的场景。

如果想要实现全局顺序消息，那么只能使用一个队列，以及单个生产者，这会严重影响性能；

![47ac0a99d3ad4cd29fbdf9fdab215edb.png][]

因此，我们常说的顺序消息通常是只的部分顺序消息，就上面的例子来说，我们不用管不同的订单ID的消息之间的总体消费顺序，只需要保证同样订单ID的消息能按照订单创建、订单付款、订单完成这个顺序消费就可以了；

**顺序消费实际上有两个核心点，一个是生产者有序存储，另一个是消费者有序消费；**

**（1）生产者有序发送**

我们知道RocketMQ中生产者生产的消息会放置在某个队列中，基于队列FIFO先进先出的特性天然的可以保证存入队列的消息顺序和拉取的消息顺序是一致的，因此，我们只需要保证一组相同的消息按照给定的顺序存**入同一个队列中**，就能保证生产者有序存储；

普通发送消息的模式下，一个topic下有多个队列，生产者会采用轮询的方式将消费均匀的分发到不同的队列中，然后被不同的消费者消费；因为一组消息在不同的队列，此时就无法使用队列天然的FIFO特性来保证消息有序性了；

RocketMQ支持生产者在投放消息的时候自定义投放策略，即为当前消息选择投放到的队列；RocketMQ提供MessageQueueSelector队列选择机制，他有三种实现：

![b3c5e3c72d564e24905331cbedb8c7e4.png][]

我们实现一个MessageQueueSelector接口（使用Hash取模法），来保证同一个订单在同一个队列中就行了，即通过订单ID%队列数量得到该ID的订单所投放的队列在队列列表中的索引，然后该订单的所有消息都会被投放到这个队列中；

另外，对于消息生产方，顺序消息必须使用同步发送的方式才能保证生产者发送的消息有序，这个也很好理解吧；

举个例子，有2个队列，那么订单ID为1,2,3的三组消息中，1、3组消息根据Hash取模结果存放于第一个队列，而2组消息存放于第二个队列，如下图是一种消息可能的消息存放顺序：

![e599447201524c6aaef4e985fd9c8db8.png][]

**注意**：实际上，采用队列选择器的方法有时也会存在问题；我们的目的是将消息发送到同一个队列中，试想如果某个broker挂了，那么队列数就会减少一部分，如果采用Hash取余的方式投递，可能导致同一个业务中的不同消息在队列数发生变化前后，分别被发送到不同的队列中，导致短暂的部分消息无序；同样的，如果增加了队列服务器数，那么也会造成短暂的造成部分消息无序；

**（2）消费者有序消费**

生产者有序存储实现了，那么该如何实现消费者有序消费呢？

RockerMQ的MessageListener回调函数提供了两种消费模式，有序消费模式MessageListenerOrderly和并发消费模式MessageListenerConcurrently；

消费者可以通过注册MessageListenerOrderly类型的回调接口来实现顺序消费；如果消费者采用Concurrently并行消费，则仍然不能保证消息消费顺序；

实际上，每一个消费者的的消费端都是采用线程池实现多线程消费的模式，即消费端是多线程消费；虽然MessageListenerOrderly被称为有序消费模式，但是仍然是使用的线程池去消费消息；

MessageListenerConcurrently是拉取到新消息之后就提交到线程池去消费，而MessageListenerOrderly则是通过**加分布式锁**和**本地锁**保证同时只有一条线程去消费一个队列上的数据；

**MessageListenerOrderly的加锁机制：**

1.  消费者在进行某个队列的消息拉取时首先向Broker服务器申请队列锁，如果申请到锁，则拉取消息，否则放弃消息拉取，等到下一个队列负载周期(20s)再试；**这个锁使得一个MessageQueue同一个时刻只能被一个消费客户端消费**；
2.  假设消费者对messageQueue的加锁已经成功，那么会开始拉取消息，拉取到消息后同样会提交到消费端的线程池进行消费；但在本地消费之前，会先获取该messageQueue对应的锁对象，每一个messageQueue对应一个锁对象，获取到锁对象后，使用synchronized阻塞式的申请线程级独占锁；**这个锁使得来自同一个messageQueue的消息在本地的同一个时刻只能被一个消费客户端中的一个线程顺序的消费**；
3.  在本地加synchronized锁成功之后，还会判断：如果是广播模式，则直接进行消费，如果是集群模式，则判断如果messagequeue没有锁住或者锁过期(默认30000ms)，那么延迟100ms后再次尝试向Broker申请锁定messageQueue，锁定成功后重新提交消费请求；

目前来说，消费者使用MessageListenerOrderly顺序消费有个两个问题：

1.  使用了很多的锁，降低了吞吐量；
2.  前一个消息消费阻塞时后面消息都会被阻塞；如果遇到消费失败的消息，会自动对当前消息进行重试（每次间隔时间为1秒），无法自动跳过，重试最大次数是Integer.MAX\_VALUE，这将导致当前队列消费暂停，因此通常需要设定有一个最大消费次数，以及处理好所有可能的异常情况；

### 3. 消息与分布式事务 ###

**什么是事务？**

事务就是一系列操作，要么同时成功，要么同时失败；事务就是为了保证一系列操作可以正常执行，它必须同时满足ACID特性；

**什么是分布式事务？**

可以想一下，你下单流程可能涉及到10多个环节，你下单付钱都成功了，但是你优惠券扣减失败了，积分新增失败了，前者公司会被薅羊毛，后者用户会不开心，但是这些都在不同的服务怎么保证大家都成功呢？——使用分布式事务；

分布式事务的实现大概分为：

*  2pc（两段式提交）
 *  3pc（三段式提交）
 *  TCC（Try、Confirm、Cancel）
 *  最大努力通知
 *  XA
 *  本地消息表（ebay研发出的）
 *  半消息/最终一致性（MQ）

分布式事务的引入是为了解决分布式环境下的某个业务场景的事务问题，当然也都有种种弊端，例如：  
（1）长时间锁定数据库资源，导致系统的响应不快，并发上不去。  
（2）网络抖动出现脑裂情况，导致事物参与者，不能很好地执行协调者的指令，导致数据不一致。  
（3）单点故障，例如事物协调者，在某一时刻宕机，虽然可以通过选举机制产生新的Leader，但是这过程中，必然出现问题；

这里介绍下最简单的2pc（两段式）方案，以及订单场景下结合MQ中间件比较常用的最终一致性方案，目的是理解下分布式事务以及其中消息中间件的作用；

**2pc（两段式提交）**

![4bbc56c73a784f9b8324cc807bc34119.jpeg][]

2pc（两段式提交）可以说是分布式事务的最开始的样子了；如图，通过消息中间件协调多个系统，在两个系统操作事务的时候都锁定资源但是不提交事务，等两者都准备好了，告诉消息中间件，然后再分别提交事务；

但是问题也很明细，如果A系统事务提交成功了，但是B系统在提交的时候网络波动或者各种原因提交失败了，其实还是会导致事务失效的；

**最终一致性**

![c4839ffddde94b708f53c760a8ef3f0f.jpeg][]

如图，整个流程中，我们能保证是：

*  业务推送方本地事务提交失败，业务接收方不会收到消息的投递；
 *  只要业务主动方本地事务执行成功，那么消息服务一定会投递消息给下游的业务被动方，并最终保证业务被动方一定能成功消费该消息（消费成功或失败，即最终一定会有一个最终态）；

> 技术就是这样，各种极端的情况我们都需要考虑，也很难有完美的方案，所以才会有这么多的方案三段式、TCC、最大努力通知等等分布式事务方案，大家只需要知道为啥要做，做了有啥好处，有啥坏处，在实际开发的时候都注意下就好好了，系统都是根据业务场景设计出来的，离开业务的技术没有意义，离开技术的业务没有底气；
> 
> ——[《敖丙的博客》][Link 4]

参考：

[《字节跳动面试官这样问消息队列：分布式事务、重复消费、顺序消费》][Link 5]

[《业务上需要顺序消费，怎么保证时序性？》][Link 6]

[《RocketMQ的顺序消息（顺序消费）》][RocketMQ]本文的图多来自这篇文章，写的很好！

[Link 1]: https://blog.csdn.net/minghao0508/article/details/125856965
[Link 2]: https://baike.baidu.com/item/%E5%B9%82%E7%AD%89/8600688?fr=aladdin
[MQ]: https://www.modb.pro/db/149777
[Link 3]: https://github.com/apache/rocketmq/blob/master/docs/cn/features.md
[47ac0a99d3ad4cd29fbdf9fdab215edb.png]: https://img-blog.csdnimg.cn/47ac0a99d3ad4cd29fbdf9fdab215edb.png
[b3c5e3c72d564e24905331cbedb8c7e4.png]: https://img-blog.csdnimg.cn/b3c5e3c72d564e24905331cbedb8c7e4.png
[e599447201524c6aaef4e985fd9c8db8.png]: https://img-blog.csdnimg.cn/e599447201524c6aaef4e985fd9c8db8.png
[4bbc56c73a784f9b8324cc807bc34119.jpeg]: https://img-blog.csdnimg.cn/4bbc56c73a784f9b8324cc807bc34119.jpeg
[c4839ffddde94b708f53c760a8ef3f0f.jpeg]: https://img-blog.csdnimg.cn/c4839ffddde94b708f53c760a8ef3f0f.jpeg
[Link 4]: https://aobing.blog.csdn.net/?type=blog
[Link 5]: https://blog.csdn.net/qq_35190492/article/details/103232854?ops_request_misc=%257B%2522request%255Fid%2522%253A%2522165814578716782395329934%2522%252C%2522scm%2522%253A%252220140713.130102334.pc%255Fblog.%2522%257D&request_id=165814578716782395329934&biz_id=0&utm_medium=distribute.pc_search_result.none-task-blog-2~blog~first_rank_ecpm_v1~rank_v31_ecpm-2-103232854-null-null.185%5Ev2%5Econtrol&utm_term=%E6%B6%88%E6%81%AF%E9%98%9F%E5%88%97&spm=1018.2226.3001.4450
[Link 6]: https://learn.lianglianglee.com/%E4%B8%93%E6%A0%8F/%E5%88%86%E5%B8%83%E5%BC%8F%E6%8A%80%E6%9C%AF%E5%8E%9F%E7%90%86%E4%B8%8E%E5%AE%9E%E6%88%9845%E8%AE%B2-%E5%AE%8C/32%20%E4%B8%9A%E5%8A%A1%E4%B8%8A%E9%9C%80%E8%A6%81%E9%A1%BA%E5%BA%8F%E6%B6%88%E8%B4%B9%EF%BC%8C%E6%80%8E%E4%B9%88%E4%BF%9D%E8%AF%81%E6%97%B6%E5%BA%8F%E6%80%A7%EF%BC%9F.md
[RocketMQ]: https://blog.csdn.net/weixin_43767015/article/details/121028059