#yyds干货盘点# 如何处理消费过程中的重复消息()

落花踏尽游何处,笑入胡姬酒肆中。这篇文章主要讲述#yyds干货盘点# 如何处理消费过程中的重复消息?相关的知识,希望能为你提供帮助。
消息传递过程中若失败,则发送方会执行重试,重试就可能产生重复消息。若不处理重复消息,可能收获惊喜。比如一个消费订单消息,统计下单金额的微服务。若不正确处理重复消息,就会出现重复统计。那仅靠MQ能保证消息不重复吗?
消息重复必然存在,在MQTT协议,给出三种传递消息时能够提供的

1 服务质量标准服务质量从低到高:
At most once
至多一次。消息在传递时,最多被送达一次。即没什么消息可靠性保证,允许丢消息。一般都是一些对消息可靠性要求不太高的监控场景使用,比如每分钟上报一次机房温度数据,可接受数据少量丢失
At least once
至少一次。消息在传递时,至少会被送达一次。即不允许丢消息,但允许少量重复消息
Exactly once
恰好一次。消息在传递时,只会被送达一次,不允许丢失也不允许重复
服务质量标准不仅适于MQTT,对所有MQ都适用。大部分MQ提供服务质量都是At least once,如RocketMQ、RabbitMQ和Kafka。可以说MQ本身并不保证消息不重复。

你不对,我看过Kafka文档,Kafka支持Exactly once的!没错,Kafka的确支持Exactly once,但本文说的也没问题。Kafka的“Exactly once”和消息传递服务质量标准中的“Exactly once”不同,它是Kafka提供的另一特性,Kafka中支持的事务也和通常理解的事务有差异。Kafka中的事务和Excactly once主要为配合流计算。
既然MQ无法保证消息不重复,就得消费代码接受“消息可能重复”这个现实,通过业务代码解决重复消息对业务的影响。

2 幂等性一般解决重复消息方案就是在消费端,让消费消息的操作具备幂等性(Idempotence):
描述一个操作、方法或者服务,其任意多次执行所产生的影响均与一次执行的影响相同。
一个幂等的方法,使用同样参数,对它进行多次调用和一次调用,对系统产生影响一样。所以,对幂等方法,无需担心重复执行会改变系统。

示例
不考虑并发,“将账户X的余额设为100元”,执行一次后对系统的影响是,账户X的余额变成了100元。只要提供参数100元不变,执行多少次,账户X余额始终100,这操作就是个幂等操作。
“将账户X余额加100元”,这操作就不是幂等,每执行次,账户余额增加100,执行多次和执行一次对系统的影响(即账户余额)不同。
若系统消费消息的业务逻辑具幂等性,那就不用担心消息重复,因为同一消息,消费一次和多次对系统影响一样。即消费多次等于消费一次。
从对系统影响结果:At least once

  • 统影响结果:At least once + 幂等消费 = Exactly once。
3 幂等实现方案最好从业务逻辑入手,将消费业务设计成具备幂等性的操作。但也不是所有业务都天然幂等,需要一些技巧。
3.1 数据库唯一约束
比如对于:将账户X余额加100。
可限制对每个转账单,每个账户只能执行一次变更操作。最简单的,在DB中建一张【转账流水表】:

  • 转账单ID
  • 账户ID
  • 变更金额
然后给【转账单ID,账户ID】联合起来创建唯一约束,这样相同转账单ID、账户ID,表里至多只存在一条记录。
消费消息逻辑可变为:“在【转账流水表】增加一条转账记录,再根据转账记录,异步更新用户余额。”
在转账流水表加条转账记录操作中,由于【转账单ID,账户ID】唯一约束,对同一转账单,同一账户只能插一条记录,后续重复插入操作都会失败,这就实现了幂等。
所以,只要是支持类似“INSERT IF NOT EXIST”语义的存储系统都可实现幂等。
比如,可用

Redis的SETNX替代数据库中的唯一约束,实现幂等消费。
3.2 为更新的数据设前置条件(类似CAS)
给数据变更设置一个前置条件:
  • 满足条件就更新数据
  • 否则拒绝更新数据
更新数据时,同时变更前置条件中需要判断的数据。于是,重复执行该操作时,由于第一次更新数据时,已变更前置条件中的判断数据,不满足前置条件,则不会再执行更新。
“将账户X的余额增加100元”,这操作加个前置条件,变为:“若账户X当前余额为500元,将余额加100元”就具备幂等性。对应到MQ消息,在消息体中带上当前余额,消费时判断DB中当前余额==消息中的余额,相等时才执行更新。
但要更新数据不是数值,或要做个复杂的更新操作咋办?前置判断条件是啥呢?

MVCC更通用的,是给数据增加版本号version属性,每次更新数据前,比较
当前数据version == 消息中的version

  • 不一致,拒绝更新
  • 一致,更新数据同时将版本号+1,一样则可实现幂等更新
3.3 记录并检查操作
若前两种方案都不适用,还有通用性最强、适用范围最广方案:记录并检查操作,也称“Token机制或GUID(全局唯一ID)机制”,执行数据更新操作前,先检查是否执行过这更新操作。
  • 发消息时,给每条消息指定全局唯一ID
  • 消费时,先根据ID检查消息是否被消费过,若没有,才更新数据并将消费状态置为已消费
但分布式系统下很难实现:
  • 首先,给每个消息指定一个全局唯一ID,方法很多,但都不太好同时满足简单、高可用和高性能,或多或少都有牺牲
  • 更麻烦的,“检查消费状态,然后更新数据并设置消费状态”,三个操作必须作为一组操作,保证原子性,才能真正实现幂等,否则就是Bug
比如对于同一消息:“全局ID为8,操作为:给ID为666账户增加100元”,可能出现这样情况:
  • t0时刻:Consumer A 收到条消息,检查消息执行状态,发现消息未处理过,开始执行“账户增加100元”
  • t1时刻:Consumer B 收到条消息,检查消息执行状态,发现消息未处理过,因这时刻,Consumer A还未来得及更新消息执行状态
  • 这样就导致账户被错误地增加了两次100元,这是一个在分布式系统中非常容易犯的错误
对此,可以用事务实现,也可以锁,但在分布式系统下,分布式事务、分布式锁都会引入高复杂度。所以一般不推荐。
总结【#yyds干货盘点# 如何处理消费过程中的重复消息()】这些幂等方案不仅可用于解决重复消息问题,也可解决重复请求或重复调用问题。比如:
  • 将HTTP服务设计成幂等的,解决前端或APP重复提交表单数据的问题
  • 将一个微服务设计成幂等的,解决RPC框架自动重试导致的重复调用问题
为何MQ都只提供At least once服务质量,而非Exactly once
若MQ实现exactly once,会引发:
  • 消费端pull时,需检测此消息是否被消费,这检测机制无疑拉低消息消费速度。随消息剧增,消费性能势必急剧下降,导致消息积压
  • 检查机制还需业务端去配合实现,若一条消息长时间未返回ack,MQ需要去回调看下消费结果(类似事务消息的回查机制)。这就增加业务端的压力与未知因素。
  • 为了确保消息没有被丢失或者重复,队列需采取一定的类似回查的手段,检测消费者是否有收到消息进行处理,在一定程度上会导致队列堆积等一系列问题,并且队列实现的复杂度上升
  • 从消费者的角度而言,因为消费者端和Broker Service端都是会各自集群,消费者端可能会存在网络抖动,导致Broker Service为了确保消息不丢失和重复,需要一直进行回查类似的操作,但是由于网络问题,导致队列堆积。
所以,MQ不实现exactly once,而是at least once
  • MQ不实现exactly once,而是at least once + 幂等性,而幂等性我们消费端业务代码自己处理。
MQ即使做到Exactly once级别,Con也要做幂等。因为Con从MQ取消息时,若Con消费成功,但ack失败,Con还是会取到重复消息,所以MQ费力做成Exactly once无法避免业务侧消息重复问题。
使用DB的唯一索引防止消息被重复消费,若业务系统存在分库分表,消费消息被路由到不同库或表,还是会存在问题?
一般也不会有问题,因为使用我们的方法,一条具体消息,总会落到确定的库表,其重复消息也会落地同样库表。
若队列实现At least once,但为不丢消息,Broker Service会进行一定重试,但不可能一直重试,若就是一直重试还是失败怎么处理?
有的MQ会有个特殊队列,保存这些总是消费失败的“坏消息”,然后继续消费之后的消息,避免这些坏消息卡死队列。这种坏消息一般不会是因为网络原因或消费者宕机导致的,大多都是因为消息数据本身有问题,消费者的业务逻辑无法处理。
exactly once,实现有性能损耗,并发高时易出现消息堆积;消息队列设计初衷是解决解耦,而解耦的对象往往是高并发,对性能要求较高的,从产品需求层面讲,消息队列设计更注重性能,而非精准(exactly once);基础架构角度来说,关注点是占比大的需求(不能不发,可以重发),占比极小的需求(敏感型,只能触发一次)可以单独抽出来另外实现。最后,请教老师有没有比较具体的业务场景,非用这种exactly once不可的

    推荐阅读