灵魂发问：重复消费顺序消费分布式事务

作者： Captain 2021-10-19 08:01:41

开发

前端

分布式 这一篇我们要说的是重复消费、顺序消费这两个在消息队列中常见的问题，以及一种事务消息，这种事务消息可以在消息队列中完成分布式事务的特性。

自流井ssl适用于网站、小程序/APP、API接口等需要进行数据传输应用场景，ssl证书未来市场广阔！成为成都创新互联公司的ssl证书销售渠道，可以享受市场价格4-6折优惠！如果有意向欢迎电话联系或者加微信：13518219792（备注：SSL证书合作）期待与您的合作！

​[[429685]]​​

hello大家好

我是大家的学习成长小伙伴Captain

我们继续学习RocketMQ，上一篇我们学习了广播消息、延迟消息、批量消息、过滤消息这些在RocketMQ中的特性，这一篇我们继续来学习RocketMQ中的那些奇奇怪怪的特性，让你在开发中如鱼得水

这一篇我们要说的是重复消费、顺序消费这两个在消息队列中常见的问题，以及一种事务消息，这种事务消息可以在消息队列中完成分布式事务的特性

把之前的这些技术点有关的文章贴到这里，大家可以先读一读

• 搞懂什么是RocketMQ
• 我怎么不知道RocketMQ生产者有这么多用法?(图片在末尾，不谢)
• 面试官问我：分布式事务是什么?

像这种啊，应该都是面试场上非Ban必选的技术点，除非面试官忘记了，否则他大概率会问起这些问题相关的技术栈，到时候可以到了发挥大家技术海和技术深度的时候了

01 重复消费问题

问题开始

我们来聊一聊消息队列中的重复消费问题吧

这种问题应该是必然存在的，也是大家使用消费队列必须考虑的问题之一，反正我用消息队列这个问题都是首先考虑的，因为这个问题如果不去考虑，可能会造成业务上的不可接受的问题

重复消费，大家肯定也明白啥意思，就是同样的消息消费了多次

为什么说这种问题必然存在呢，因为消息队列一定有它的重试机制，也就是消息重发，一旦消费端出现异常的情况下，消息队列会进行消息的重发

你重发消息重新处理没问题，但是一般一个消息的监听者不止一个，也就是可能多个系统都在监听着处理这个消息，别的系统要是不支持重复消费，那岂不很糟糕

别的系统的数据就会出现混乱，各个系统之间的数据便会出现不一致的情况

举个例子，电商系统中的支付成功消息，支付成功之后发送一个消息，积分系统、物流系统多个系统监听这一消息，积分系统处理出现异常，该支付成功的消息重新发送了一条，物流系统要是不支持消息的重试，那就出现了两个物流单子，那可能会造成客户买了一件商品，付了一件商品的钱，结果呢，给用户发了多个该商品

​​​

啊这...

这样岂不糟糕透了?你也可能该收拾东西了

其实出现消息重试这真的真的是很常见的情况，也是大家在使用消息队列必须必须要考虑的，比如网络抖动、系统业务的处理bug等，这个问题不处理，系统后患无穷

那这种重复消费问题如何避免呢

解决方案：幂等

​​​

简单来说，幂等是一个数学上的概念，通俗的解释就是同样的参数多次调用同样的接口，调用的结果都是一样的，也就是你支付成功的消息发送多少次，最终生成的物流数据还是一条

这样就没问题了

那如何去保证幂等呢

像这种问题我一般是分为两种场景去回答的，一种是生产端的幂等，另一种是消费端的幂等

生产者端的幂等一般都是通过第三方的存储来完成的，比如Redis，或者是流水表，在消息发送之后，将记录暂时保存起来，下次发送消息之前，在Redis中检查该消息是否发送过，不过这种在很多场景下是不合适的，这种会在生产端就限制了重试这一机制

如果生产端发送成功，消费失败，则不会重新发送该消息

另一种消费者端的幂等，这种是属于最常见的，生产者无论发送多少次同样的消息，最终的执行结果都是一样的，可以分为强幂等和弱幂等来处理

强幂等其实就是用于必须幂等的业务场景，不允许出现差错的，这种更为谨慎些，比如上面的支付成功的这种消息，物流消费方的处理肯定要是强幂等咯

这里可以引进一个三方存储，流水表或者Redis都可以，支付成功之后，记录到流水表中，这里用Redis可能会丢失，把支付成功和记录到流水表放入到同一个事务中，要么一起成功，要么一起失败

每次消息过来之后根据订单号去流水表中检查是否有这条流水，有流水则直接return就可以了

也可以直接用数据库的唯一约束来做insert操作

还以一种属于是弱幂等性，这种不能保证百分百情况下幂等，比如用Redis来存储业务ID作为唯一key来处理，Redis宕机可能导致短信发送情况的丢失，不过问题不大，用户也是可以接受的，我们来看一下实例代码

 String idempotentValue = RedisUtil.get(RedisConstant.IDEMPOTENT.concat(msgId), String.class);   if (!StringUtils.isEmpty(idempotentValue)) {       log.info("========该消息已经被消费：【{}】", msgBody);       return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;     }  //业务代码 //幂等处理  RedisUtil.setEx(RedisConstant.IDEMPOTENT.concat(msgId), "1", 5, TimeUnit.DAYS); 

02 顺序消费

那你说一下你有没有遇到过顺序消费这个场景呢?

顺序消费这个场景其实不是特别的常见，但是也是必不可少的，因为在某些业务场景下顺序是很关键的，保证消息的消费顺序也是很关键的

比如我们有一个操作需要对数据进行删除、增加、修改三个操作，这种在一般的系统中我们都会采用SQL来进行操作，但是当数据量很大的时候，我们做备份同步数据的时候，这种同步有的时候会通过消息队列来慢慢的去执行，这个时候就很有必要保证消息的顺序性，如果上面的三个操作变成了修改、删除、增加这样的顺序，那就不是我们想要的效果了

普通的消息的消费当然是没有固定顺序的，消息发送的时候默认是采用的轮询的方式发送到不同的分区中

​​​

而消费端消费的时候则是会分配到多个分区的，多个分区是同时拉取提交消费的，在同一个分区queue中，是可以保证FIFO的，但是普通消息是没法达到顺序消费的，只需要将消息投递到同一条queue中即可

​​​

按照上面所说，我们只需要保证需要保持顺序的消息投递到相同的queue中即可，这样同一个queue中的消息肯定会投递到同一个消费实例，同一个消费实例肯定是顺序拉取消息，然后顺序的去消费

即使触发重排导致queue分配给了别的消费者也没有关系，由于queue的消息永远是FIFO的，所以只需要保证消息的重复消费的幂等性即可，queue的内部顺序还是没问题的

顺序消费分配全局顺序和分区顺序

• 全局顺序：对于指定的一个Topic，所有消息按照严格的先入先出FIFO(First In First Out)的顺序进行发布和消费。
• 分区顺序：对于指定的一个Topic，所有消息根据Sharding Key进行区块分区。同一个分区内的消息按照严格的FIFO顺序进行发布和消费。Sharding Key是顺序消息中用来区分不同分区的关键字段，和普通消息的Key是完全不同的概念。

为什么全局魂虚顺序消息消费性能一般

全局顺序消息是严格按照FIFO的消息阻塞原则，即上一条消息没有被成功消费，那么下一条消息会一直被存储到Topic队列中。如果想提高全局顺序消息的TPS，可以升级实例配置，同时消息客户端应用尽量减少处理本地业务逻辑的耗时。

在rocketmq中，一个topic下有多个队列queue，于是乎为了保证消息的顺序性，将消息发送到同一个queue中，rocketmq提供了MessageQueueSelector队列选择机制，有三种实现

​​​

使用Hash取模法让需要顺序消费的消息发送到同一个queue中，再使用同步发送，当然这个取模根据的是这些消息的共同属性

rocketmq仅仅保证了发送的顺序性，至于最终的顺序消费还是要由消费者业务来保证，就是我保证我发给你的是按照顺序的消息，但是你要是自己给处理乱了就不关我rocketmq的事了，那就是你自己的代码问题了

其实还是存在一些异常的场景会导致出现乱序的情况，比如master宕机，导致写入队列的数量发生了变化，你想啊，采用上面的hash取模就会出现消息分散到其它的queue中，这样就不能保证有序了，除非选择master如果挂了就无法发送接下来的消息

03 分布式事务

聊一下分布式事务吧

大家看一下这篇面试官问我：分布式事务是什么?

简单来说就是，事务是要么全部执行成功，要么全部执行失败;而分布式事务就是跨机器的，跨服务的，跨系统的事务保证，现在的系统都是拆分成很多的服务，每个服务最少部署两台，分别部署在不同的机器上

这样系统之间的事务保证就是分布式事务

而rocketmq中的事务消息则天然支持分布式事务

事务消息：实现类似X或者Open XA的分布式事务功能，以达到最终一致性

消息队列RocketMQ版提供类似X或Open XA的分布式事务功能，通过消息队列RocketMQ版事务消息，能达到分布式事务的最终一致。

半事务消息：暂不能投递的消息，发送方已经成功地将消息发送到了消息队列RocketMQ版服务端，但是服务端未收到生产者对该消息的二次确认，此时该消息被标记成“暂不能投递”状态，处于该种状态下的消息即半事务消息。

消息回查：由于网络闪断、生产者应用重启等原因，导致某条事务消息的二次确认丢失，消息队列RocketMQ版服务端通过扫描发现某条消息长期处于“半事务消息”时，需要主动向消息生产者询问该消息的最终状态(Commit或是Rollback)，该询问过程即消息回查。

​​​

跟Captain来看看事务消息发送步骤：

1、发送方将半事务消息发送到服务端Broker，服务端会将消息持久化，成功之后会返回ACK确认消息已经发送成功，此时消息为半事务消息

2、发送方开始执行本地事务的逻辑

3、发送方会根据本地事务的执行结果向服务端提交二次确认，决定Commit还是Rollback，服务端收到Commit之后则把这个消息标记为可投递，发送到消费方;服务端收到Rollback之后则删除半事务消息，服务端不会发送，则消费方也不会收到

如可是如果断网或者应用重启这些情况，上述的步骤的二次确认信息无法到达服务端，怎么办?

这里其实有个回查机制，发送方发送消息之后，需要本地执行事务，如果事务执行的过程出现卡死的情况，或者事务执行结果因为网络等问题，无法传递事务结果到服务端，服务端会执行一个回查机制，来确认这个半事务消息的最终提交情况

本文转载自微信公众号「Java贼船」

​​​

文章题目：灵魂发问：重复消费顺序消费分布式事务
网站网址：http://www.shufengxianlan.com/qtweb/news39/59239.html

网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 创新互联