大厂都是如何处理重复消息的?

2022-09-14 08:02:20 浏览数 (1)

消息消费失败,很多框架会自动执行重试,而重试就产生了重复消息。

MQTT协议给出三种传递消息时能够提供的

1 服务质量(Quality of Service)

从低到高:

1.1 QoS 0:At most once

消息最多传递一次,如果当时客户端不可用,则会丢失该消息。即消息在传递时,最多被送达一次。无消息可靠性保证,允许丢消息。

一种 “fire and forget” 的消息发送模式:Sender (Publisher 或 Broker) 发送一条消息之后,就不再关心它有没有发送到对方,也不设置任何重发机制。

消息分发依赖于底层网络能力。发布者只会发布一次消息,接收者不会应答消息,发布者也不会储存和重发消息。该等级具有最高传输效率,但可能送达一次也可能根本没送达。

一般都是一些对消息可靠性要求不太高的监控场景使用,如每s上报一次司机乘客地理位置,可接受数据少量丢失。

1.2 QoS 1:At least once

消息传递至少 1 次。消息在传递时,至少会被送达一次。即不允许丢消息,但允许重复消息。

包含简单的重发机制,Sender 发送消息之后等待接收者的 ACK,若没收到 ACK,则重发消息。这种模式能保证消息至少能到达一次,但无法保证消息重复。

MQTT 通过简单的 ACK 机制保证 QoS 1。发布者会发布消息,并等待接收者的 PUBACK 报文的应答,若规定时间内没收到 PUBACK 应答,发布者会将消息的 DUP 置为 1 并重发。接收者接收到 QoS 为 1 的消息时应该回应 PUBACK 报文,接收者可能会多次接受同一个消息,无论 DUP 标志如何,接收者都会将收到的消息当作一个新的消息并发送 PUBACK 报文应答。

1.3 QoS 2:Exactly once

恰好一次。消息在传递时,只会被送达一次,不允许丢失、重复。设计了重发和重复消息发现机制,保证消息到达对方并且严格只到达一次。最高等级服务质量,消息丢失和重复都不可接受。使用该等级有额外开销。

发布者发布 QoS 为 2 的消息之后,会将发布的消息储存起来并等待接收者回复 PUBREC 的消息,发送者收到 PUBREC 消息后,它就可以安全丢弃掉之前的发布消息,因为它已经知道接收者成功收到了消息。发布者会保存 PUBREC 消息并应答一个 PUBREL,等待接收者回复 PUBCOMP 消息,当发送者收到 PUBCOMP 消息之后会清空之前所保存的状态。

当接收者接收到一条 QoS 为 2 的 PUBLISH 消息时,他会处理此消息并返回一条 PUBREC 进行应答。当接收者收到 PUBREL 消息之后,它会丢弃掉所有已保存的状态,并回复 PUBCOMP。

无论在传输过程中何时出现丢包,发送端都负责重发上一条消息。不管发送端是 Publisher 还是 Broker,都是如此。因此,接收端也需要对每一条命令消息都进行应答。

1.4 QoS 在发布与订阅中的区别

MQTT 发布与订阅操作中的 QoS 代表不同含义:

  • 发布时的 QoS,消息发送到服务端时使用的 QoS
  • 订阅时的 QoS,服务端向自己转发消息时可使用的最大 QoS
  • 当客户端 A 的发布 QoS 大于客户端 B 的订阅 QoS 时,服务端向客户端 B 转发消息时使用的 QoS 为客户端 B 的订阅 QoS。
  • 当客户端 A 的发布 QoS 小于客户端 B 的订阅 QoS 时,服务端向客户端 B 转发消息时使用的 QoS 为客户端 A 的发布 QoS。

不同情况下客户端收到的消息 QoS 可参考下表:

1.5 QoS 等级选型

QoS 级别越高,流程越复杂,系统资源消耗越大。应用程序可以根据自己的网络场景和业务需求,选择合适级别。

QoS 0
  • 可以接受消息偶尔丢失。
  • 在同一个子网内部的服务间的消息交互,或其他客户端与服务端网络非常稳定的场景。
QoS 1
  • 对系统资源消耗较为关注,希望性能最优化。
  • 消息不能丢失,但能接受并处理重复的消息。
QoS 2
  • 不能忍受消息丢失(消息的丢失会造成生命或财产的损失),且不希望收到重复的消息。
  • 数据完整性与及时性要求较高的银行、消防、航空等行业。

大部分MQ都是At least once,如RocketMQ、RabbitMQ和Kafka,即MQ本身并不保证消息不重复。

1.6 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 幂等消费 = Exactly once。

3 幂等实现方案

最好从业务逻辑入手,将消费业务设计成具备幂等性的操作。但也不是所有业务都天然幂等,需要一些技巧。

3.1 数据库唯一约束

比如对于:将账户X余额加100。

可限制对每个转账单,每个账户只能执行一次变更操作。最简单的,在DB中建一张【转账流水表】:

  • 转账单ID
  • 账户ID
  • 变更金额

然后给【转账单ID,账户ID】联合起来创建唯一约束,这样相同转账单ID、账户ID,表里至多只存在一条记录。

消费消息逻辑可变为:“在【转账流水表】增加一条转账记录,再根据转账记录,异步更新用户余额。” 在转账流水表加条转账记录操作中,由于【转账单ID,账户ID】唯一约束,对同一转账单,同一账户只能插一条记录,后续重复插入操作都会失败,这就实现了幂等。

所以,只要是支持类似“INSERT IF NOT EXIST”语义的存储系统都可实现幂等。 比如,可用

Redis的SETNX

替代数据库中的唯一约束,实现幂等消费。

该种方案需要消费者基于消息类型,去感知此消息类型所要处理的业务,在业务上的唯一约束,不同业务的唯一约束不一样,对消费者实现幂等不友好。

但解决不了主动的重试问题吧,比如插入流水,执行业务,返回MQ逻辑错误,触发重新消费,这时会发现流水已存在。所以这里插流水和业务逻辑也得在一个事务里,这跟方法按区别看来只是怎么去控制唯一性而已。只要流水正确写入了,后续根据流水计算余额的业务逻辑可不与写入流水在同一个事务,即使计算余额失败,也能根据流水重新计算。

3.2 为更新的数据设前置条件(类似CAS)

给数据变更设置一个前置条件:

  • 满足条件就更新数据
  • 否则拒绝更新数据

更新数据时,同时变更前置条件中需要判断的数据。于是,重复执行该操作时,由于第一次更新数据时,已变更前置条件中的判断数据,不满足前置条件,则不会再执行更新。

“将账户X的余额增加100元”,这操作加个前置条件,变为:“若账户X当前余额为500元,将余额加100元”就具备幂等性。对应到MQ消息,在消息体中带上当前余额,消费时判断DB中当前余额==消息中的余额,相等时才执行更新。

但要更新数据不是数值,或要做个复杂的更新操作咋办?前置判断条件是啥呢?

当余额为500时,执行加100,若当前消息被消费前,下一条消息到来时,数据库余额还是500,这时设置更新条件也是500,这种问题怎么解决? 这种场景就得保证消息的严格顺序。

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元,这是一个在分布式系统中非常容易犯的错误

对此,可以用事务实现,也可以锁,但在分布式系统下,分布式事务、分布式锁都会引入高复杂度。所以一般不推荐。

由生产者将不同业务的不同唯一约束(如A业务是a b字段须唯一,B业务是a c字段须唯一),统一处理成对消费者友好的全局唯一ID,如A业务是md5(a b),B业务是md5(a c),生成全局唯一ID,可以是上面举例的本地md5计算,也可以是包装成服务接口,但其本身也必须幂等,如此Con不管处理什么业务消息,都只需针对"全局唯一ID"保证幂等。

4 总结

这些幂等方案不仅可用于解决重复消息问题,也可解决重复请求或重复调用问题。比如:

  • 将HTTP服务设计成幂等的,解决前端或APP重复提交表单数据的问题
  • 将一个微服务设计成幂等的,解决RPC框架自动重试导致的重复调用问题

4.1 为何MQ只提供At least once,而非Exactly once

若MQ实现exactly once,会引发:

  • 消费端pull时,需检测此消息是否被消费,这检测机制无疑拉低消息消费速度。随消息剧增,消费性能势必急剧下降,导致消息积压
  • 检查机制还需业务端去配合实现,若一条消息长时间未返回ack,MQ需要去回调看下消费结果(类似事务消息的回查机制)。这就增加业务端的压力与未知因素。
  • 为了确保消息没有被丢失或者重复,队列需采取一定的类似回查的手段,检测消费者是否有收到消息进行处理,在一定程度上会导致队列堆积等一系列问题,并且队列实现的复杂度上升
  • 从消费者的角度而言,因为消费者端和Broker Service端都是会各自集群,消费者端可能会存在网络抖动,导致Broker Service为了确保消息不丢失和重复,需要一直进行回查类似的操作,但是由于网络问题,导致队列堆积

exactly once实现有性能损耗,并发高时易出现消息堆积;消息队列设计初衷是解决解耦,而解耦的对象往往是高并发,对性能要求较高的:

  • 从产品需求层面讲,MQ设计更注重性能,而非精准(exactly once)
  • 基础架构角度来说,关注点是占比大的需求(不能不发,但可以重发),占比极小的需求(敏感型,只能触发一次)可单独抽出来另外实现

所以,MQ不实现exactly once,而是at least once 幂等性,而幂等性我们消费端业务代码自己处理。

MQ即使做到Exactly once级别,Con也要做幂等。因为Con从MQ取消息时,若Con消费成功,但ack失败,Con还是会取到重复消息,所以MQ费力做成Exactly once无法避免业务侧消息重复问题。

4.2 使用DB的唯一索引防止消息被重复消费,若业务系统存在分库分表,消费消息被路由到不同库或表,还是会存在问题?

一般也不会有问题,因为使用我们的方法,一条具体消息,总会落到确定的库表,其重复消息也会落地同样库表。

4.3 若队列实现At least once,为了不丢消息,Broker Service会进行一定重试,但不可能一直重试,若就是一直重试还是失败怎么处理?

rabbitmq有个特殊队列保存这些总是消费失败的“坏消息”,然后继续消费之后的消息,避免这些坏消息卡死队列。这种坏消息一般不是因为网络原因或消费者宕机导致,大多都是因为消息数据本身有问题,消费者的业务逻辑无法处理。

只支持At least once:是不是与以下几种情况相关: 1.硬件异常或者系统异常导致的数据丢失:消息队列为何不能做成像数据库一样的用undo log和redo log去避免硬件的这种异常,出于性能考虑

为何网络协议中一样TCP和UDP的区别:消息反馈可能不是每一个反馈一次,有时是一批反馈异常,传输中可能会出现丢包或者顺序不一致。 大部分MQ都是批量收发,但采用基于位置的确认机制,可保证顺序。

kafka就算用事务,也不能保证没有重复消费,它有可能发生rebalance时,消费了数据没有提交

关于幂等的情况,像设置帐户余额为100元,或者给余额为500的加100,如果有中间状态的变更或者ABA问题,也能算是幂等操作吗? 确实这个例子解决不了ABA问题,如果要解决这个问题,只能使用版本号方式。

因为目前消息队列,在发送消息给客户端的时候,一般需要客户端ack之后才能确定,这条消息是不是真的被消费了:

  • 如果客户端设置的是自动ack,那么mq就能保证只发送一次,但是这样会因为客户端消费消息不成功,而导致消息丢失
  • 如果客户端都设置手动ack,若MQ发消息给客户端成功,客户端也消费完成,就在准备ack时,和MQ失去联系,这时MQ不知道这条消息是否真的被消费,只能选择重发消息

所以若MQ保证了只发一次,则MQ就无法保证消息由于客户端消费失败而不丢失,就好像分布式系统中的cap理论,只能保证其中的两种,而无法三个都保证。架构设计就是在取舍之间选择最合适的实现方式。

“如果账户 X 当前的余额为 500 元,将余额加 100 元"和“检查消息执行状态,发现消息未处理过,开始执行账户增加 100”,这两者有啥区别,不都是消费端compareAndUpdate吗,都可以用普通数据库事务就能实现。 主要是检查的内容不一样:

  • 前者检查余额,容易实现,但适用范围比较窄
  • 后者检查消息执行状态,难实现,但适用范围更广泛
  • 如何解决方案一和方案二日益增多的存储日志呀,有合适的删除策略吗? 这种流水一般不能删除,若数量太多影响查询消息,可考虑按照账户ID来分表存储。

参考

  • [MQTT QoS(服务质量)介绍 | EMQ (emqx.com)](

0 人点赞