RocketMQ(七)——Transaction Message(事务消息)

分布式事务

通过MQ解决分布式事务的思路

1) 业务和消息生成耦合在一起

2) 业务和消息解耦

RocketMQ 中的事务消息

1) 目前RMQ3.2.6中事务消息的实现原理及存在的问题

2) 问题解决思路

本文介绍RocketMQ提供的第三种类型的消息——Transaction Message(事务消息)。在说事务消息之前,我们先来说说分布式事务的那些事!

分布式事务

什么是分布式事务,我的理解是一半事务。怎么说,比如有2个异构系统,A异构系统要做T1,B异构系统要做T2,要么都成功,要么都失败。
要知道异构系统,很显然,不在一个数据库实例上,它们往往分布在不同物理节点上,本地事务已经失效

 2阶段提交

2阶段提交协议,Two-Phase Commit,是处理分布式事务的一种常见手段。2PC,存在2个重要角色:事务协调器(TC),事务执行者。
2PC,可以看到节点之间的通信次数太多了,时间很长!时间变长了,从而导致,事务锁定的资源时间也变长了,造成资源等待时间变长!在高并发场景下,存在严重的性能问题!

通过MQ解决分布式事务的思路

下面,我们来看看MQ在高并发场景下,是如何解决分布式事务的。

考虑生活中的场景:

我们去北京庆丰包子铺吃炒肝,先去营业员那里付款(Action1),拿到小票(Ticket),然后去取餐窗口排队拿炒肝(Action2)。思考2个问题:第一,为什么不在付款的同时,给顾客炒肝?如果这样的话,会增加处理时间,使得后面的顾客等待时间变长,相当于降低了接待顾客的能力(降低了系统的QPS)。第二,付了款,拿到的是Ticket,顾客为什么会接受?从心理上说,顾客相信Ticket会兑现炒肝。事实上也是如此,就算在最后炒肝没了,或者断电断水(系统出现异常),顾客依然可以通过Ticket进行退款操作,这样都不会有什么损失!(虽然这么说,但是实际上包子铺最大化了它的利益,如果炒肝真的没了,浪费了顾客的时间,不过顾客顶多发发牢骚,最后接受)

生活已经告诉我们处理分布式事务,保证数据最终一致性的思路!这个Ticket(凭证)其实就是消息!

1) 业务和消息生成耦合在一起

业务和消息生成耦合在一起

业务操作和消息的生成耦合在一起,保证了只要A银行的账户发生扣款,那么一定会生成一条转账消息。只要A银行系统的事务成功提交,我们可以通过实时消息服务,将转账消息通知B银行系统,如果B银行系统回复成功,那么A银行系统可以在table中设置这条转账消息的状态。
这样耦合的方式,从架构上来看,就有点不太优雅,而且存在一些问题。比如说,消息的存储实质上是在A银行系统中的,如果A银行系统出了问题,将导致无法转账。如果解耦,将消息独立出来呢?

2) 业务和消息解耦

如上图所示,消息数据独立存储,业务和消息解耦,实质上消息的发送有2次,一条是转账消息,另一条是确认消息。

RocketMQ 中的事务消息

1) 目前RMQ3.2.6中事务消息的实现原理及存在的问题

到这里,先来看看基于RocketMQ的代码:

生产者示例代码
生产者这里用到是:TransactionMQProducer。
这里涉及到2个角色:本地事务执行器(代码中的TransactionExecuterImpl)、服务器回查客户端Listener(代码中的TransactionCheckListener)。如果事务消息发送到MQ上后,会回调  本地事务执行器;但是此时事务消息是prepare状态,对消费者还不可见,需要  本地事务执行器  返回RMQ一个确认消息。

本地事务执行器

事务消息是否对消费者可见,完全由事务返回给RMQ的状态码决定(状态码的本质也是一条消息)。

回查Listener

生产者发送了2条消息给RMQ,有一条本地事务执行成功,有一条本地事务执行失败。总共是2条业务消息 + 2条确认消息,因此是4条。注意到消费者只消费了一条数据,就是只有告诉RMQ本地事务执行成功的那条消息才会被消费!因此是1条!

但是,注意到本地事务执行失败的消息,RMQ并没有check listener?

这是为什么呢?因为RMQ在3.0.8的时候还是支持check listener回查机制的,但是到了3.2.6的时候将事务回查机制“阉割”了!

那么3.0.8的时候,RMQ是怎么做事务回查的呢?

看一看源码,你会知道,其实事务消息开始是prepare状态,然后RMQ会将其持久化到MySQL当中,然后如果收到确认消息,就删除掉这条prepare消息,如果迟迟收不到确认消息,那么RMQ会定时的扫描prepare消息,发送给produce group进行回查确认!

到这里,问题来了,要知道3.2.6版本,没有回查机制了,会存在问题么?

当然会存在问题!假设,我们发送一条转账事务消息给RMQ,成功后回调本地事务,DB减操作成功,刚准备给RMQ一个确认消息,此时突然断电,或者网络抖动,使得这条确认消息没有发送出去。此时RMQ中的那条转账事务消息,始终处于prepare状态,消费者读取不到,但是却已经完成一方的账户资金变动!!!

2) 问题解决思路

既然,RMQ3.2.6版本不为我们进行回查,那么只能由我们自己完成了。
重新看一下“业务和消息解耦”的那张转账流程的图:

在正常情况下,当然没有问题,如果第五步(向MQ发送确认消息)出现失败,加上RocketMQ 3.2.6版本没有事务回查机制,就会导致这条转账消息,在A银行完成了操作,但是迟迟对B银行系统不可见!

用户U1从A银行系统转账给B银行系统的用户U2的处理过程如下:

A银行系统生成一条转账消息,以事务消息的方式写入RocketMQ,此时B银行系统不可见这条消息

写入MQ成功后,回调A银行系统,对T1,T2表进行操作(很显然需要是一个事务)

我们重点关注下T2表,这个表是用来干嘛的呢?每条转账消息都会在T2表中,该表有2个特殊的字段:status,updatetime。(用途会在后文详述)
完成第二步,接下来发送确认消息给MQ,如果这个确认消息发送成功,那么这条转账消息,将对B银行系统可见。然后B银行系统,会在一个事务中完成对t3,t5的操作。

如果发送确认消息给MQ失败的处理思路:

首先,B银行系统,有一个定时任务(比如说每隔1MIN执行一次),扫描表t5,取得一段时间内的数据,发送给A银行系统。要知道t5中的数据,必然是A银行系统成功处理并发送确认消息成功的转账数据。为什么要发送给A银行系统呢,其实就是为了找到那些发送确认消息失败的转账数据。那么怎么发给A银行系统呢,这个方式比较多,可以考虑在来一个Topic,也可以考虑Netty等。发送给A银行系统,其实就是为了更新t2表的status,updatetime。

这里有一个关键,如何“扫描表t5,取得一段时间内的数据”?这就是t4的作用,在t4中记录一个time字段,每次定时任务启动,先更新time(比如设定为当前系统时间,设置前的的时间为old),然后扫描出t5中大于这个old时间的转账数据,如此循环往复。

其次,A银行系统,也有一个定时任务(可以根据业务消费能力定,可以大一些),扫描t2表(指定status及updatetime条件),将那些确认消息发送失败的转账消息找出来,更新updatetime并发送给MQ。

这样,我们并没有改动RocketMQ 3.2.6的源码,而是在外围解决了事务回查!

其实到这里,你可以发现RocketMQ的一个特点,就是将生产者和MQ绑定,而不需要特别处理消费者,这是为什么呢?

因为消息只要发往RocketMQ成功,那么就意味着成功,为什么这么说?前面,我们说过,消费者端消费消息只会产生2种错误,第一:timeout,第二:exception。要知道RocketMQ对于超时,会不断重试;对于消费异常,会根据消费端的返回码,会有重试机制保证。也就是,RocketMQ一定会让消息得到消费,如果消费有问题,只能是消费者的问题,而不会是RocketMQ的问题!

我这儿整理了比较全面的JAVA相关的面试资料,


需要领取面试资料的同学,请加群:473984645

原文地址:https://www.cnblogs.com/1013wang/p/12200724.html

时间: 2024-08-01 14:45:42

RocketMQ(七)——Transaction Message(事务消息)的相关文章

Apache RocketMQ 正式开源分布式事务消息

近日,Apache RocketMQ 社区正式发布4.3版本.此次发布不仅包括提升性能,减少内存使用等原有特性增强,还修复了部分社区提出的若干问题,更重要的是该版本开源了社区最为关心的分布式事务消息,而且实现了对外部组件的零依赖.接下来,本文将详细探秘RocketMQ事务消息的设计原理以及实现机制. 一.需求缘起 在微服务架构中,随着服务的逐步拆分,数据库私有已经成为共识,这也导致所面临的分布式事务问题成为微服务落地过程中一个非常难以逾越的障碍,但是目前尚没有一个完整通用的解决方案. 其实不仅仅

RocketMQ 整合SpringBoot发送事务消息

环境 jdk: 8u22rocketmq: rocketmq-all-4.5.2-bin-releasespringboot: 2.1.6.RELEASErocketmq-springboot: 2.0.3 发送流程(事务消息) Rocket发送事务消息:1.由producer发送prepare(半消息)给MQ的broker2.prepare消息发送成功以后执行本地业务(本地事务),根据本地事务执行结果手动返回相应状态(RocketMQLocalTransactionState.COMMIT.R

RocketMQ实现事务消息方案

RocketMQ 是一个来自阿里巴巴的分布式消息中间件,于 2012 年开源,并在 2017 年正式成为 Apache 顶级项目.据了解,包括阿里云上的消息产品以及收购的子公司在内,阿里集团的消息产品全线都运行在 RocketMQ 之上,并且最近几年的双十一大促中,RocketMQ 都有抢眼表现.Apache RocketMQ 4.3之后的版本正式支持事务消息,为分布式事务实现提供了便利性支持. RocketMQ 事务消息设计则主要是为了解决 Producer 端的消息发送与本地事务执行的原子性

RocketMQ事务消息学习及刨坑过程

一.背景 MQ组件是系统架构里必不可少的一门利器,设计层面可以降低系统耦合度,高并发场景又可以起到削峰填谷的作用,从单体应用到集群部署方案,再到现在的微服务架构,MQ凭借其优秀的性能和高可靠性,得到了广泛的认可. 随着数据量增多,系统压力变大,开始出现这种现象:数据库已经更新了,但消息没发出来,或者消息先发了,但后来数据库更新失败了,结果研发童鞋各种数据修复,这种生产问题出现的概率不大,但让人很郁闷.这个其实就是数据库事务与MQ消息的一致性问题,简单来讲,数据库的事务跟普通MQ消息发送无法直接绑

消息中间件 RocketMQ源码解析:事务消息

关注微信公众号:[芋艿的后端小屋]有福利: RocketMQ / MyCAT / Sharding-JDBC 所有源码分析文章列表 RocketMQ / MyCAT / Sharding-JDBC 中文注释源码 GitHub 地址 您对于源码的疑问每条留言都将得到认真回复.甚至不知道如何读源码也可以请教噢. 新的源码解析文章实时收到通知.每周更新一篇左右. 1. 概述 2. 事务消息发送 2.1 Producer 发送事务消息 2.2 Broker 处理结束事务请求 2.3 Broker 生成

RocketMQ源码分析之从官方示例窥探:RocketMQ事务消息实现基本思想

RocketMQ4.3.0版本开始支持事务消息,后续分享将开始将剖析事务消息的实现原理.首先从官方给出的Demo实例入手,以此通往RocketMQ事务消息的世界中. 官方版本未发布之前,从apache rocketmq第一个版本上线后,代码中存在与事务消息相关的代码,例如COMMIT.ROLLBACK.PREPARED,在事务消息未开源之前网上对于事务消息的"声音"基本上是使用类似二阶段提交,主要是根据消息系统标志MessageSysFlag中定义来推测的: TRANSACTION_P

RocketMQ源码分析之RocketMQ事务消息实现原理中篇----事务消息状态回查

上节已经梳理了RocketMQ发送事务消息的流程(基于二阶段提交),本节将继续深入学习事务状态消息回查,我们知道,第一次提交到消息服务器时消息的主题被替换为RMQ_SYS_TRANS_HALF_TOPIC,本地事务执行完后如果返回本地事务状态为UN_KNOW时,第二次提交到服务器时将不会做任何操作,也就是说此时消息还存在与RMQ_SYS_TRANS_HALF_TOPIC主题中,并不能被消息消费者消费,那这些消息最终如何被提交或回滚呢? 原来RocketMQ使用TransactionalMessa

搞懂分布式技术19:使用RocketMQ事务消息解决分布式事务

搞懂分布式技术19:使用RocketMQ事务消息解决分布式事务 初步认识RocketMQ的核心模块 rocketmq模块 rocketmq-broker:接受生产者发来的消息并存储(通过调用rocketmq-store),消费者从这里取得消息. rocketmq-client:提供发送.接受消息的客户端API. rocketmq-namesrv:NameServer,类似于Zookeeper,这里保存着消息的TopicName,队列等运行时的元信息.(有点NameNode的味道) rocketm

RocketMQ 事务消息

RocketMQ 事务消息在实现上充分利用了 RocketMQ 本身机制,在实现零依赖的基础上,同样实现了高性能.可扩展.全异步等一系列特性. 在具体实现上,RocketMQ 通过使用 Half Topic 以及 Operation Topic 两个内部队列来存储事务消息推进状态,如下图所示: 其中,Half Topic 对应队列中存放着 prepare 消息,Operation Topic 对应的队列则存放了 prepare message 对应的 commit/rollback 消息,消息体