消息队列队列：重复消费，顺序消费分布式事务

Posted 2021-02-28 qingaoaoo

1、消息队列的消息重复消费

消息重复消费是使用消息队列之后，必须考虑的一个问题，也是比较严重和常见的问题，但凡用到了消息队列，第一时间考虑的就是重复消费的问题。

就比如有这样的一个场景，用户下单成功后我需要去一个活动页面给他加GMV（销售总额），最后根据他的GMV去给他发奖励，这是电商活动很常见的玩法。类似累计下单金额到哪个梯度给你返回什么梯度的奖励这样。

这样的活动页面10000%是用异步去加的，而且大家应该也有这样的体会，你下单了马上去看一些活动页面，有时候马上就有了，有时候却延迟有很久，为啥？这个速度取决于消息队列的消费速度，消费慢堵塞了就迟点看到。

你下个单支付成功你就发个消息出去，我们上面那个活动的开发人员就监听你的支付成功消息，我监听到你这个订单成功支付的消息，那我就去我活动GMV表里给你加上去，听到这里大家可能觉得顺理成章。

但是我告诉大家一般消息队列的使用，我们都是有重试机制的，就是说我下游的业务发生异常了，我会抛出异常并且要求你重新发一次。但是不止你一个人监听这个消息啊，还有别的服务也在监听，他们也会失败啊，他一失败他也要求重发，但是你这里其实是成功的，重发了，你的钱不就加了两次了？

就好比上面的这样，我们的积分系统处理失败了，他这个系统肯定要求你重新发送一次这个消息对吧，积分的系统重新接收并且处理成功了，但是别人的活动，优惠券等等服务也监听了这个消息呀，那不就可能出现活动系统给他加GMV加两次，优惠券扣两次这种情况么？

真实的情况其实重试是很正常的，服务的网络抖动，开发人员代码Bug，还有数据问题等都可能处理失败要求重发的。

开发过程中是怎么去保证重复消费的问题

一般我们叫这样的处理叫接口幂等。

幂等（idempotent、idempotence）是一个数学与计算机学概念，常见于抽象代数中。

在编程中一个幂等操作的特点是其任意多次执行所产生的影响均与一次执行的影响相同。

幂等函数，或幂等方法，是指可以使用相同参数重复执行，并能获得相同结果的函数。这些函数不会影响系统状态，也不用担心重复执行会对系统造成改变。

例如，“setTrue()”函数就是一个幂等函数,无论多次执行，其结果都是一样的.更复杂的操作幂等保证是利用唯一交易号(流水号)实现.

一般幂等，分场景去考虑，看是强校验还是弱校验，比如跟金钱相关的场景那就很关键，就做强校验，别不是很重要的场景做弱校验。

强校验：

比如你监听到用户支付成功的消息，你监听到了去加GMV是不是要调用加钱的接口，那加钱接口下面再调用一个加流水的接口，两个放在一个事务，成功一起成功失败一起失败。

每次消息过来都要拿着订单号+业务场景这样的唯一标识（比如天猫双十一活动）去流水表查，看看有没有这条流水，有就直接return不要走下面的流程了，没有就执行后面的逻辑。

之所以用流水表，是因为涉及到金钱这样的活动，有啥问题后面也可以去流水表对账，还有就是帮助开发人员定位问题。

 

弱校验：

这个简单，一些不重要的场景，比如给谁发短信啥的，我就把这个id+场景唯一标识作为Redis的key，放到缓存里面失效时间看你场景，一定时间内的这个消息就去Redis判断。

用KV就算消息丢了可能这样的场景也没关系，反正丢条无关痛痒的通知短信嘛（你敢说你没验证码短信丢失的情况？）。

还有很多公司的弱校验用token啊什么的，反正花样很多，但是重要的场景一定要强校验，真正查问题的时候没有在磁盘持久化的数据，心里还是空空的，就像你和女朋友分开的时候的心里状态一样。（我单身的怎么知道这种感觉？猜的）

2、消息顺序消费的场景

Tip：一般都是同个业务场景下不同几个操作的消息同时过去，本身顺序是对的，但是你发出去的时候同时发出去了，消费的时候却乱掉了，这样就有问题了。在数据量大的时候数据同步压力还是很大的，有时候数据量大的表需要同步几个亿的数据。（并不是主从同步，主从延迟大的话会有问题，可能是从数据库或者主数据库同步到备库）

这种情况我们都是怼到队列里面去，然后慢慢消费的，那问题就来了呀，我们在数据库同时对一个Id的数据进行了增、改、删三个操作，但是你消息发过去消费的时候变成了改，删、增，这样数据就不对了。

本来一条数据应该删掉了，结果在你那却还在，这不是出大问题！

两者的结果是不是完全不一样了 ↑

怎么解决呢？

我简单的说一下我们使用的RocketMQ里面的一个简单实现吧。

Tip：为啥用RocketMQ举例呢，这玩意是阿里开源的，我问了下身边的朋友很多公司都有使用，所以读者大概率是这个的话我就用这个举例吧，具体的细节我后面会在RocketMQ和Kafka各自章节说到。

生产者消费者一般需要保证顺序消息的话，可能就是一个业务场景下的，比如订单的创建、支付、发货、收货。

那这些东西是不是一个订单号呢？一个订单的肯定是一个订单号的说，那简单了呀。

一个topic下有多个队列，为了保证发送有序，RocketMQ提供了MessageQueueSelector队列选择机制，他有三种实现:

我们可使用Hash取模法，让同一个订单发送到同一个队列中，再使用同步发送，只有同个订单的创建消息发送成功，再发送支付消息。这样，我们保证了发送有序。

RocketMQ的topic内的队列机制,可以保证存储满足FIFO（First Input First Output 简单说就是指先进先出）,剩下的只需要消费者顺序消费即可。

RocketMQ仅保证顺序发送，顺序消费由消费者业务保证!!!

这里很好理解，一个订单你发送的时候放到一个队列里面去，你同一个的订单号Hash一下是不是还是一样的结果，那肯定是一个消费者消费，那顺序是不是就保证了？

真正的顺序消费不同的中间件都有自己的不同实现我这里就举个例子，大家思路理解下。

Tip：我写到这点的时候人才群里也有人问我，一个队列有序出去，一个消费者消费不就好了，我想说的是消费者是多线程的，你消息是有序的给他的，你能保证他是有序的处理的？还是一个消费成功了再发下一个稳妥。

3、分布式事务么？

分布式事务在现在遍地都是分布式部署的系统中几乎是必要的。

我们先聊一下啥是事务？

分布式事务、事务隔离级别、ACID我相信大家这些东西都耳熟能详了，那什么是事务呢？

概念：

一般是指要做的或所做的事情。

在计算机术语中是指访问并可能更新数据库中各种数据项的一个程序执行单元(unit)。

事务通常由高级数据库操纵语言或编程语言（如SQL，C++或Java）书写的用户程序用户程序的执行所引起，并用形如begin transaction和end transaction语句（或函数调用）来界定。

事务由事务开始(begin transaction)和事务结束(end transaction)之间执行的全体操作组成。

特性：

事务是恢复和并发控制的基本单位。

事务应该具有4个属性：原子性、一致性、隔离性、持久性。这四个属性通常称为ACID特性。

原子性（atomicity）：一个事务是一个不可分割的工作单位，事务中包括的操作要么都做，要么都不做。

一致性（consistency）：事务必须是使数据库从一个一致性状态变到另一个一致性状态。一致性与原子性是密切相关的。

隔离性（isolation）：一个事务的执行不能被其他事务干扰。即一个事务内部的操作及使用的数据对并发的其他事务是隔离的，并发执行的各个事务之间不能互相干扰。

持久性（durability）：持久性也称永久性（permanence），指一个事务一旦提交，它对数据库中数据的改变就应该是永久性的。接下来的其他操作或故障不应该对其有任何影响。

那有同学还是不理解，敖丙我总结了一下就是：事务就是一系列操作，要么同时成功，要么同时失败。然后会从事务的 ACID 特性（原子性、一致性、隔离性、持久性）展开叙述。

事务就是为了保证一系列操作可以正常执行，它必须同时满足 ACID 特性。

 

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

聪明，分布式事务，你看你都会抢答了！

Tip：真实的应用场景可能比我介绍的场景复杂数倍，我只是为了举例方便一下大家理解所以用了很简单的例子。

我接触和了解到的分布式事务大概分为：

• 2pc（两段式提交）

• 3pc（三段式提交）

• TCC（Try、Confirm、Cancel）

• 最大努力通知

• XA

• 本地消息表（ebay研发出的）

• 半消息/最终一致性（RocketMQ）

这里我就介绍下最简单的2pc（两段式），以及大家以后可能比较常用的半消息事务也就是最终一致性，目的是让大家理解下分布式事务里面消息中间件的作用，别的事务都大同小异，都有很多优点。

当然也都有种种弊端：

例如长时间锁定数据库资源，导致系统的响应不快，并发上不去。

网络抖动出现脑裂情况，导致事物参与者，不能很好地执行协调者的指令，导致数据不一致。

单点故障：例如事物协调者，在某一时刻宕机，虽然可以通过选举机制产生新的Leader，但是这过程中，必然出现问题，而TCC，只有强悍的技术团队，才能支持开发，成本太高。

不多BB了，我们开始介绍这个两个事物吧。

2pc（两段式提交） :

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

但是我不知道大家看到问题所在没有？

是的你可能已经发现了，如果A系统事务提交成功了，但是B系统在提交的时候网络波动或者各种原因提交失败了，其实还是会失败的。

最终一致性：

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

• 业务主动方本地事务提交失败，业务被动方不会收到消息的投递。

• 只要业务主动方本地事务执行成功，那么消息服务一定会投递消息给下游的业务被动方，并最终保证业务被动方一定能成功消费该消息（消费成功或失败，即最终一定会有一个最终态）。

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

还是那句话：没有最完美的系统，只有最适合的系统。