如何处理消费过程中的重复消息？

如何处理消费过程中的重复消息？

消息传递过程中若失败，则发送方会执行重试，重试就可能产生重复消息。若不处理重复消息，可能收获惊喜。比如一个消费订单消息，统计下单金额的微服务。若不正确处理重复消息，就会出现重复统计。那仅靠MQ能保证消息不重复吗？

消息重复必然存在，在MQTT协议，给出三种传递消息时能够提供的

1 服务质量标准

服务质量从低到高：

At most once

至多一次。消息在传递时，最多被送达一次。即没什么消息可靠性保证，允许丢消息。一般都是一些对消息可靠性要求不太高的监控场景使用，比如每分钟上报一次机房温度数据，可接受数据少量丢失

At least once

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

Exactly once

恰好一次。消息在传递时，只会被送达一次，不允许丢失也不允许重复

服务质量标准不仅适于MQTT，对所有MQ都适用。大部分MQ提供服务质量都是At least once，如RocketMQ、RabbitMQ和Kafka。可以说MQ本身并不保证消息不重复。

没错，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”语义的存储系统都可实现幂等。

比如，可用

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

3.2 为更新的数据设前置条件（类似CAS）

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

满足条件就更新数据

否则拒绝更新数据

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

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

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

更通用的，是给数据增加版本号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元，这是一个在分布式系统中非常容易犯的错误

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

总结

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

将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级别，Con也要做幂等。因为Con从MQ取消息时，若Con消费成功，但ack失败，Con还是会取到重复消息，所以MQ费力做成Exactly once无法避免业务侧消息重复问题。

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

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

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

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

exactly once，实现有性能损耗，并发高时易出现消息堆积；消息队列设计初衷是解决解耦，而解耦的对象往往是高并发，对性能要求较高的，从产品需求层面讲，消息队列设计更注重性能，而非精准（exactly once）；基础架构角度来说，关注点是占比大的需求（不能不发，可以重发），占比极小的需求（敏感型，只能触发一次）可以单独抽出来另外实现。最后，请教老师有没有比较具体的业务场景，非用这种exactly once不可的

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。