消息幂等(去重)通用解决方案,真顶!

作者 | 薛定谔的风口猪

来源 | https://jaskey.github.io/blog/2020/06/08/rocketmq-message-dedup/

消息中间件是分布式系统常用的组件,无论是异步化、解耦、削峰等都有广泛的应用价值。我们通常会认为,消息中间件是一个可靠的组件——这里所谓的可靠是指,只要我把消息成功投递到了消息中间件,消息就不会丢失,即消息肯定会至少保证消息能被消费者成功消费一次,这是消息中间件最基本的特性之一,也就是我们常说的“AT LEAST ONCE”,即消息至少会被“成功消费一遍”。

举个例子,一个消息M发送到了消息中间件,消息投递到了消费程序A,A接受到了消息,然后进行消费,但在消费到一半的时候程序重启了,这时候这个消息并没有标记为消费成功,这个消息还会继续投递给这个消费者,直到其消费成功了,消息中间件才会停止投递。

然而这种可靠的特性导致,消息可能被多次地投递。举个例子,还是刚刚这个例子,程序A接受到这个消息M并完成消费逻辑之后,正想通知消息中间件“我已经消费成功了”的时候,程序就重启了,那么对于消息中间件来说,这个消息并没有成功消费过,所以他还会继续投递。这时候对于应用程序A来说,看起来就是这个消息明明消费成功了,但是消息中间件还在重复投递。

这在RockectMQ的场景来看,就是同一个messageId的消息重复投递下来了。

基于消息的投递可靠(消息不丢)是优先级更高的,所以消息不重的任务就会转移到应用程序自我实现,这也是为什么RocketMQ的文档里强调的,消费逻辑需要自我实现幂等。背后的逻辑其实就是:不丢和不重是矛盾的(在分布式场景下),但消息重复是有解决方案的,而消息丢失是很麻烦的。

简单的消息去重解决方案

例如:假设我们业务的消息消费逻辑是:插入某张订单表的数据,然后更新库存:

insert into t_order values .....
update t_inv set count = count-1 where good_id = 'good123';

要实现消息的幂等,我们可能会采取这样的方案:

select * from t_order where order_no = 'order123'if(order  != null) {return ;//消息重复,直接返回}

这对于很多情况下,的确能起到不错的效果,但是在并发场景下,还是会有问题。

并发重复消息

假设这个消费的所有代码加起来需要1秒,有重复的消息在这1秒内(假设100毫秒)内到达(例如生产者快速重发,Broker重启等),那么很可能,上面去重代码里面会发现,数据依然是空的(因为上一条消息还没消费完,还没成功更新订单状态),

那么就会穿透掉检查的挡板,最后导致重复的消息消费逻辑进入到非幂等安全的业务代码中,从而引发重复消费的问题(如主键冲突抛出异常、库存被重复扣减而没释放等)

并发去重的解决方案之一

要解决上面并发场景下的消息幂等问题,一个可取的方案是开启事务把select 改成 select for update语句,把记录进行锁定。

select * from t_order where order_no = 'THIS_ORDER_NO' for update  //开启事务
if(order.status != null) {return ;//消息重复,直接返回
}

但这样消费的逻辑会因为引入了事务包裹而导致整个消息消费可能变长,并发度下降。

当然还有其他更高级的解决方案,例如更新订单状态采取乐观锁,更新失败则消息重新消费之类的。但这需要针对具体业务场景做更复杂和细致的代码开发、库表设计,不在本文讨论的范围。

但无论是select for update, 还是乐观锁这种解决方案,实际上都是基于业务表本身做去重,这无疑增加了业务开发的复杂度, 一个业务系统里面很大部分的请求处理都是依赖MQ的,如果每个消费逻辑本身都需要基于业务本身而做去重/幂等的开发的话,这是繁琐的工作量。本文希望探索出一个通用的消息幂等处理的方法,从而抽象出一定的工具类用以适用各个业务场景。

Exactly Once

在消息中间件里,有一个投递语义的概念,而这个语义里有一个叫”Exactly Once”,即消息肯定会被成功消费,并且只会被消费一次。以下是阿里云里对Exactly Once的解释:

Exactly-Once 是指发送到消息系统的消息只能被消费端处理且仅处理一次,即使生产端重试消息发送导致某消息重复投递,该消息在消费端也只被消费一次。

在我们业务消息幂等处理的领域内,可以认为业务消息的代码肯定会被执行,并且只被执行一次,那么我们可以认为是Exactly Once。

但这在分布式的场景下想找一个通用的方案几乎是不可能的。不过如果是针对基于数据库事务的消费逻辑,实际上是可行的。

基于关系数据库事务插入消息表

假设我们业务的消息消费逻辑是:更新MySQL数据库的某张订单表的状态:

update t_order set status = 'SUCCESS' where order_no= 'order123';

要实现Exaclty Once即这个消息只被消费一次(并且肯定要保证能消费一次),我们可以这样做:在这个数据库中增加一个消息消费记录表,把消息插入到这个表,并且把原来的订单更新和这个插入的动作放到同一个事务中一起提交,就能保证消息只会被消费一遍了。

  1. 开启事务

  2. 插入消息表(处理好主键冲突的问题)

  3. 更新订单表(原消费逻辑)

  4. 提交事务

说明:

  1. 这时候如果消息消费成功并且事务提交了,那么消息表就插入成功了,这时候就算RocketMQ还没有收到消费位点的更新再次投递,也会插入消息失败而视为已经消费过,后续就直接更新消费位点了。这保证我们消费代码只会执行一次。

  2. 如果事务提交之前服务挂了(例如重启),对于本地事务并没有执行所以订单没有更新,消息表也没插入成功;而对于RocketMQ服务端来说,消费位点也没更新,所以消息还会继续投递下来,投递下来发现这个消息插入消息表也是成功的,所以可以继续消费。这保证了消息不丢失。

事实上,阿里云ONS的EXACTLY-ONCE语义的实现上,就是类似这个方案基于数据库的事务特性实现的。更多详情可参考:https://help.aliyun.com/document_detail/102777.html

基于这种方式,的确这是有能力拓展到不同的应用场景,因为他的实现方案与具体业务本身无关——而是依赖一个消息表。

但是这里有它的局限性

  1. 消息的消费逻辑必须是依赖于关系型数据库事务。如果消费的消费过程中还涉及其他数据的修改,例如Redis这种不支持事务特性的数据源,则这些数据是不可回滚的。

  2. 数据库的数据必须是在一个库,跨库无法解决

注:业务上,消息表的设计不应该以消息ID作为标识,而应该以业务的业务主键作为标识更为合理,以应对生产者的重发。阿里云上的消息去重只是RocketMQ的messageId,在生产者因为某些原因手动重发(例如上游针对一个交易重复请求了)的场景下起不到去重/幂等的效果(因消息id不同)。

更复杂的业务场景

如上所述,这种方式Exactly Once语义的实现,实际上有很多局限性,这种局限性使得这个方案基本不具备广泛应用的价值。并且由于基于事务,可能导致锁表时间过长等性能问题。

例如我们以一个比较常见的一个订单申请的消息来举例,可能有以下几步(以下统称为步骤X):

  1. 检查库存(RPC)

  2. 锁库存(RPC)

  3. 开启事务,插入订单表(MySQL)

  4. 调用某些其他下游服务(RPC)

  5. 更新订单状态

  6. commit 事务(MySQL)

这种情况下,我们如果采取消息表+本地事务的实现方式,消息消费过程中很多子过程是不支持回滚的,也就是说就算我们加了事务,实际上这背后的操作并不是原子性的。怎么说呢,就是说有可能第一条小在经历了第二步锁库存的时候,服务重启了,这时候实际上库存是已经在另外的服务里被锁定了,这并不能被回滚。当然消息还会再次投递下来,要保证消息能至少消费一遍,换句话说,锁库存的这个RPC接口本身依旧要支持“幂等”。

再者,如果在这个比较耗时的长链条场景下加入事务的包裹,将大大的降低系统的并发。所以通常情况下,我们处理这种场景的消息去重的方法还是会使用一开始说的业务自己实现去重逻辑的方式,如前面加select for update,或者使用乐观锁。

那我们有没有方法抽取出一个公共的解决方案,能兼顾去重、通用、高性能呢?

拆解消息执行过程

其中一个思路是把上面的几步,拆解成几个不同的子消息,例如:

  1. 库存系统消费A:检查库存并做锁库存,发送消息B给订单服务

  2. 订单系统消费消息B:插入订单表(MySQL),发送消息C给自己(下游系统)消费

  3. 下游系统消费消息C:处理部分逻辑,发送消息D给订单系统

  4. 订单系统消费消息D:更新订单状态

注:上述步骤需要保证本地事务和消息是一个事务的(至少是最终一致性的),这其中涉及到分布式事务消息相关的话题,不在本文论述。

可以看到这样的处理方法会使得每一步的操作都比较原子,而原子则意味着是小事务,小事务则意味着使用消息表+事务的方案显得可行。

然而,这太复杂了!这把一个本来连续的代码逻辑割裂成多个系统多次消息交互!那还不如业务代码层面上加锁实现呢。

更通用的解决方案

上面消息表+本地事务的方案之所以有其局限性和并发的短板,究其根本是因为它依赖于关系型数据库的事务,且必须要把事务包裹于整个消息消费的环节。

如果我们能不依赖事务而实现消息的去重,那么方案就能推广到更复杂的场景例如:RPC、跨库等。

例如,我们依旧使用消息表,但是不依赖事务,而是针对消息表增加消费状态,是否可以解决问题呢?

基于消息幂等表的非事务方案

以上是去事务化后的消息幂等方案的流程,可以看到,此方案是无事务的,而是针对消息表本身做了状态的区分:消费中、消费完成。只有消费完成的消息才会被幂等处理掉。而对于已有消费中的消息,后面重复的消息会触发延迟消费(在RocketMQ的场景下即发送到RETRY TOPIC),之所以触发延迟消费是为了控制并发场景下,第二条消息在第一条消息没完成的过程中,去控制消息不丢(如果直接幂等,那么会丢失消息(同一个消息id的话),因为上一条消息如果没有消费完成的时候,第二条消息你已经告诉broker成功了,那么第一条消息这时候失败broker也不会重新投递了)

上面的流程不再细说,后文有github源码的地址,读者可以参考源码的实现,这里我们回头看看我们一开始想解决的问题是否解决了:

  1. 消息已经消费成功了,第二条消息将被直接幂等处理掉(消费成功)。

  2. 并发场景下的消息,依旧能满足不会出现消息重复,即穿透幂等挡板的问题。

  3. 支持上游业务生产者重发的业务重复的消息幂等问题。

关于第一个问题已经很明显已经解决了,在此就不讨论了。

关于第二个问题是如何解决的?主要是依靠插入消息表的这个动作做控制的,假设我们用MySQL作为消息表的存储媒介(设置消息的唯一ID为主键),那么插入的动作只有一条消息会成功,后面的消息插入会由于主键冲突而失败,走向延迟消费的分支,然后后面延迟消费的时候就会变成上面第一个场景的问题。

关于第三个问题,只要我们设计去重的消息键让其支持业务的主键(例如订单号、请求流水号等),而不仅仅是messageId即可。所以也不是问题。

此方案是否有消息丢失的风险?

如果细心的读者可能会发现这里实际上是有逻辑漏洞的,问题出在上面聊到的个三问题中的第2个问题(并发场景),在并发场景下我们依赖于消息状态是做并发控制使得第2条消息重复的消息会不断延迟消费(重试)。但如果这时候第1条消息也由于一些异常原因(例如机器重启了、外部异常导致消费失败)没有成功消费成功呢?也就是说这时候延迟消费实际上每次下来看到的都是消费中的状态,最后消费就会被视为消费失败而被投递到死信Topic中(RocketMQ默认可以重复消费16次)。

有这种顾虑是正确的!对于此,我们解决的方法是,插入的消息表必须要带一个最长消费过期时间,例如10分钟,意思是如果一个消息处于消费中超过10分钟,就需要从消息表中删除(需要程序自行实现)。所以最后这个消息的流程会是这样的:

更灵活的消息表存储媒介

我们这个方案实际上没有事务的,只需要一个存储的中心媒介,那么自然我们可以选择更灵活的存储媒介,例如Redis。使用Redis有两个好处:

  1. 性能上损耗更低

  2. 上面我们讲到的超时时间可以直接利用Redis本身的ttl实现

当然Redis存储的数据可靠性、一致性等方面是不如MySQL的,需要用户自己取舍。

源码:RocketMQDedupListener

以上方案针对RocketMQ的Java实现已经开源放到Github中,具体的使用文档可以参考https://github.com/Jaskey/RocketMQDedupListener ,

以下仅贴一个Readme中利用Redis去重的使用样例,用以意业务中如果使用此工具加入消息去重幂等的是多么简单:

//利用Redis做幂等表
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("TEST-APP1");
consumer.subscribe("TEST-TOPIC", "*");String appName = consumer.getConsumerGroup();// 大部分情况下可直接使用consumer group名
StringRedisTemplate stringRedisTemplate = null;// 这里省略获取StringRedisTemplate的过程
DedupConfig dedupConfig = DedupConfig.enableDedupConsumeConfig(appName, stringRedisTemplate);
DedupConcurrentListener messageListener = new SampleListener(dedupConfig);consumer.registerMessageListener(messageListener);
consumer.start();

以上代码大部分是原始RocketMQ的必须代码,唯一需要修改的仅仅是创建一个DedupConcurrentListener示例,在这个示例中指明你的消费逻辑和去重的业务键(默认是messageId)。

更多使用详情请参考Github上的说明。

这种实现是否一劳永逸?

实现到这里,似乎方案挺完美的,所有的消息都能快速的接入去重,且与具体业务实现也完全解耦。那么这样是否就完美的完成去重的所有任务呢?

很可惜,其实不是的。原因很简单:因为要保证消息至少被成功消费一遍,那么消息就有机会消费到一半的时候失败触发消息重试的可能。还是以上面的订单流程X:

  1. 检查库存(RPC)

  2. 锁库存(RPC)

  3. 开启事务,插入订单表(MySQL)

  4. 调用某些其他下游服务(RPC)

  5. 更新订单状态

  6. commit 事务(MySQL)

当消息消费到步骤3的时候,我们假设MySQL异常导致失败了,触发消息重试。因为在重试前我们会删除幂等表的记录,所以消息重试的时候就会重新进入消费代码,那么步骤1和步骤2就会重新再执行一遍。如果步骤2本身不是幂等的,那么这个业务消息消费依旧没有做好完整的幂等处理。

本实现方式的价值?

那么既然这个并不能完整的完成消息幂等,还有什么价值呢?价值可就大了!虽然这不是解决消息幂等的银弹(事实上,软件工程领域里基本没有银弹),但是他能以便捷的手段解决:

1.各种由于Broker、负载均衡等原因导致的消息重投递的重复问题

2.各种上游生产者导致的业务级别消息重复问题

3.重复消息并发消费的控制窗口问题,就算重复,重复也不可能同一时间进入消费逻辑

一些其他的消息去重的建议

也就是说,使用这个方法能保证正常的消费逻辑场景下(无异常,无异常退出),消息的幂等工作全部都能解决,无论是业务重复,还是rocketmq特性带来的重复。

事实上,这已经能解决99%的消息重复问题了,毕竟异常的场景肯定是少数的。那么如果希望异常场景下也能处理好幂等的问题,可以做以下工作降低问题率:

  1. 消息消费失败做好回滚处理。如果消息消费失败本身是带回滚机制的,那么消息重试自然就没有副作用了。

  2. 消费者做好优雅退出处理。这是为了尽可能避免消息消费到一半程序退出导致的消息重试。

  3. 一些无法做到幂等的操作,至少要做到终止消费并告警。例如锁库存的操作,如果统一的业务流水锁成功了一次库存,再触发锁库存,如果做不到幂等的处理,至少要做到消息消费触发异常(例如主键冲突导致消费异常等)

  4. 在#3做好的前提下,做好消息的消费监控,发现消息重试不断失败的时候,手动做好#1的回滚,使得下次重试消费成功。

往期推荐

最强代码生成器平台,杀疯了!

Spring Boot 2.x基础教程:使用Elastic Job实现定时任务

Spring为什么建议使用构造器来注入?

K8S面试大通关!赶紧收藏!

Redis 内存压缩实战,学习了!

喜欢本文欢迎转发,关注我订阅更多精彩

关注我回复「加群」,加入Spring技术交流群

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.rhkb.cn/news/21640.html

如若内容造成侵权/违法违规/事实不符,请联系长河编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

TP6关联统计无法去重的解决方案

在tp6中官方手册跟源码中是没办法做到去重关联统计的,因为源代码中的withCount方法,默认是以“*”进行统计。当我们想要统计类型有多少,而不是数量有多少的时候就会非常麻烦。 这时候需要修改源码,将默认的“*”,改成传…

postgreSql根据多个字段去重详细流程

postgreSql根据多个字段去重详细流程 这是测试表的建标语句,需要的可以拿去用。 CREATE TABLE "public"."class_name" ("id" int4 NOT NULL DEFAULT nextval(id::regclass),"class_no" varchar(50) COLLATE "pg_cat…

mysql去重函数:DISTINCT ,与GROUP_CONCAT结合去重

mysql去重函数:DISTINCT ,与GROUP_CONCAT结合 创建表语法distinct多字段去重DISTINCT 结合 group 统计实例 创建表 DROP TABLE IF EXISTS qipa_shop; CREATE TABLE qipa_shop (price varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci N…

tp6 count distinct统计时根据字段去重

代码: AppointmentModel::field([identity])->count(distinct identity)生成的sql: SELECT COUNT(DISTINCT identity) AS think_count FROM dg_appointment;下面是tp6手册内容 我尝试着这样写 发现生成的sql不对 AppointmentModel::distinct(tru…

oracle listagg如何去重

listagg去重 去重思路:利用listagg会忽略null值的特点 按ENTITY_GROUP_RRN 分组,用 listagg 分别合并 EQPT_ID 与 STATION_ID ,同时要求去重 表 T_TEST 数据如下: EQPT_IDENTITY_GROUP_RRNSTATION_IDTOOL-00110493721JITAI-1TO…

【数据去重】海量数据实时去重方案

文章目录 Prologue布隆过滤器去重什么是布隆过滤器实现的核心思想怎么理解 内嵌RocksDB状态后端去重引入外部K-V存储去重 Prologue 数据去重(data deduplication)是我们大数据攻城狮司空见惯的问题了。除了统计UV等传统用法之外,去重的意义更…

listagg结果去重

最近在一个项目中用到了listagg方法,但是在组合结果中出现有重复的情况。默认的结果如下 于是我就写了一个方法对listagg的结果去重,也可以对该格式的字符串去重,方法如下 create or replace function listaggpure(targetStr varchar2,seper…

非常实用的5种json数组去重方法,函数实现思路竟是chatgpt帮我写的!

文章目录 人工智能福利文章方法一 reduce实现思路 方法二 filter和Map实现思路 方法三 变异for实现思路 方法四 for filter实现思路 方法五 for实现思路 写在最后 人工智能福利文章 【分享几个国内免费可用的ChatGPT镜像】【10几个类ChatGPT国内AI大模型】【用《文心一言》1分…

AI自动播——AI虚拟主播帮你实现24小时直播带货技术分享

如今很多淘宝天猫商家会在做直播的时候,会遇到真人主播难招或者真人主播直播时长有限,那么有什么办法解决这个难题呢,这个时候就可以用AI自动播了,也就是AI虚拟主播直播带货。具体怎么操作呢?下面一起来看看吧。 AI自动…

亚马逊账号被关联能申诉得回来吗

关于亚马逊账号被关联能申诉回来吗?小编有话要说。 其实小编是一个特别谨慎的人,在初入亚马逊之前,会了解到亚马逊所有的问题之后,才会注册账号。那么要了解的有什么呢?注册需要的信息,注册的规则&#xf…

社区1月热门文章、ChatGPT工具汇总,强化学习安全和隐私、斯坦福大学CS234《智源社区强化学习周刊》第75期...

No.75 智源社区 强化学习组 强 化 学 习 研究 观点 资源 活动 社区热门文章 Google发布Bard与ChatGPT竞争 https://hub.baai.ac.cn/view/23925 纽约时报:ChatGPT诞生背后的故事 https://hub.baai.ac.cn/view/23835 ChatGPT提示与AI工具、开发工具、低代码工具汇总 《…

5款超实用的免费AI工具,让你轻松提升工作效率!

目录 1、AiChat 产品特点 任意选择内置角色对话 自定义AI角色进行对话 推荐理由🌟🌟🌟🌟🌟 2、MINISTER AI 产品特点 集成ChatGPT聊天 简化Stable Diffusion 集成Midjourney模式 训练并分享模型 推荐理由&a…

13 款炫酷的 MySQL 可视化管理工具!好用到爆!!

MySQL 的管理维护工具非常多,除了系统自带的命令行管理工具之外,还有许多其他的图形化管理工具,工具好用是一方面,个人的使用习惯也很重要,这里介绍 13 款 MySQL 图形化管理工具,供大家参考。 1、DBeaver D…

GPT模型支持下的Python-GEE遥感云大数据分析、管理与可视化技术及多领域案例应用

查看原文>>>GPT模型支持下的Python-GEE遥感云大数据分析、管理与可视化技术及多领域案例应用 目录 第一章、理论基础 第二章、开发环境搭建 第三章、遥感大数据处理基础与ChatGPT等AI模型交互 第四章、典型案例操作实践 第五章、输入输出及数据资产高效管理 第…

【IC基础】集成电路设计领域术语缩写及名词解释(字母索引版)

前言: 笔者在大三上学习学习《SoC设计导论》时整理的有关集成电路设计领域的常见有英文缩写和对应的名词解释,文中标注的页码均出自《SoC设计方法与实现》这本参考书: 目录 目录 前言: 目录 A B C CTG(Clock Tree Generat…

浏览器扩展合集来啦!实用而有趣的浏览器扩展插件!

五彩:网页划线、批注、高亮工具 五彩严格意义上的同类产品是 Hypothesis、diigo 等工具,Cubox、Readwise、简悦则属于需要先解析一下网页后高亮的插件。 此前我已经在用 Cubox 了,为什么还需要五彩来完成网页高亮? Cubox 在移动…

GPT模型结合Python-GEE遥感云大数据分析、管理与可视化技术

GPT模型支持下的Python-GEE遥感云大数据分析、管理与可视化技术及多领域案例应用 随着航空、航天、近地空间等多个遥感平台的不断发展,近年来遥感技术突飞猛进。由此,遥感数据的空间、时间、光谱分辨率不断提高,数据量也大幅增长&#xff0c…

分享6个AI绘画网站

ChatGPT狂飙160天,世界已经不是之前的样子。 新建了人工智能中文站https://ai.weoknow.com 每天给大家更新可用的国内可用chatGPT资源 1、Midjourney 特点:业内标杆,效果最强大 Midjourney是基于diffusion的AI图画艺术生成器。生成图片不局…

又又又发现了一个 AI 插件神器 TeamSmart

简介 TeamSmart AI 是最近比较火的 Chrome 插件,它是基于 ChatGPT 集成的 AI 助手团队工具 对,没错,是一个团队。这个团队里面有许多不同角色的成员,每隔成员都有自己的专业领域,比如商业、市场营销、灵魂写手、程序…

忘记了谷歌Gmail账号名怎么办?用这种方法轻松找回谷歌邮箱地址

有些朋友以前注册过谷歌邮箱,但很久很久没有再去使用。 现在注册ChatGPT需要谷歌邮箱,于是打算把尘封已久的谷歌邮箱找出来,可是这时候你才发现,谷歌邮箱的账号名早已忘掉了。 今天重点来说说如何找回谷歌账号,希望能够…