事务消息应用场景、实现原理与项目实战(附全部源码)

本文涉及的产品
服务治理 MSE Sentinel/OpenSergo,Agent数量 不受限
简介: 从应用场景出发,给出解决方案与实现原理,并提供整套工业级实现源码。

作者:丁威

活动中心场景介绍


在电商系统上线初期,往往会进行一些“拉新”活动,例如活动部门提出新用户注册送积分、送优惠券活动。


基于分布式、微服务的设计理念,通常的架构设计(子系统交互)如下图所示:


image.png

其核心系统介绍如下:


  • 账户中心
    提供用户登录、用户注册等服务,一个新用户注册时,向 MQ 服务器中的 USER_REGISTER 主题发送一条消息,主流程结束,与送积分,送优惠券等过程解耦。
  • 优惠券(券系统)
    提供发放优惠券、使用优惠券等与券相关的基础服务。
  • 积分中心
    提供积分相关的服务,例如积分赠送、积分消费、积分查询等基础服务。
  • 送积分服务(消费者)
    订阅 MQ,按照规则决定是否需要赠送积分,如果需要则调用积分相关的基础接口,完成积分的发放。
  • 送优惠券(消费者)
    订阅 MQ,按照规则决定是否需要赠送优惠券,如果需要则调用券系统相关的基础接口,完成优惠券的发放。


上面的架构设计非常优雅,但并不是无懈可击,如果新用户注册成功,但消息发送到 MQ 失败,或者消息成功发送到 MQ,但发送完 MQ 后系统出现异常导致用户注册失败又该如何呢?


上面的问题其实就是典型的分布式事务问题:即如何保证用户注册(数据库操作)与 MQ 消息发送这两个分布式操作的一致性。


RocketMQ 事务消息闪亮登场。


事务消息实现原理


一言以蔽之:RocketMQ 事务消息要解决的问题是消息发送与业务的一致性,其解决思路:二阶段提交与事务状态回查,其具体实现流程如下图所示:


image.png


其核心设计理念:


  • 应用程序开启一个数据库事务,进行数据库操作,并且在事务中发送一条 PREPARE 消息,PREPARE 消息发送成功后通知应用程序记录本地事务状态,然后提交本地事务。
  • RocketMQ 在收到类型为 PREPARE 的消息时,首先备份消息的原主题与原消息消费队列,然后将消息存储在主题为 RMQ_SYS_TRANS_HALF_TOPIC 的消息队列中,故 PREPARE 的消息是不会被客户端消费的。
  • Broker 消息服务器开启一个定时任务处理 RMQ_SYS_TRANS_HALF_TOPIC 中的消息,会每隔指定时间向消息发送者发起事务状态查询请求 ,询问消息发送者客户端本地事务是否成功,然后根据回查状态决定是提交还是回滚,即对处于 PREPARE 状态进行提交或回滚操作。
  • 发送者如果明确得知事务成功,则可以返回 COMMIT,服务端会提交该条消息,具体操作是恢复原消息的主题与队列,重新发送到 Broker,消费端感知后消费。
  • 发送者如果无法明确得知事务状态,则返回 UNOWN,此时服务端会等待一定时间后再次向发送者询问,默认询问 15 次。
  • 发送者如果非常明确得知事务失败,则可以返回 ROLLBACK。

在具体实践中,消息发送者在无法获取事务状态时不要武断的返回 ROLLBACK,而是要返回 UNOWN,让服务端定时重试回查,说明如下:


image.png


在将 PREPARE 消息发送到 Broker 后,服务端发起事务查询时本地事务可能还未提交,为了避免无效的事务回查机制,RocketMQ 通常至少在收到 PREPARE 消息 6s 后才会发起第一次事务回查,可通过 transactionTimeOut 配置。故客户端在实现事务回查时无法证明事务状态时不应该返回 ROLLBACK,而是返回 UNOWN。


事务消息实战


光说不练假把式,接下来以一个新用户注册送优惠券的场景来详细介绍如何使用事务消息。


项目模块职责说明如下:


image.png


事务消息的核心代码组装在 transaction-service,其核心类图如下:


image.png


其中核心要点如下:


  • UserServiceImpl
    Dubbo 接口业务实现类,类似 MVC 的控制层,在这里做一些参数验证,但不执行具体的业务逻辑,只是发送一条事务消息到 MQ
  • UserRegTransactionListener
    事务监听器,在 executeLocalTransaction 方法中执行业务逻辑,数据库本地事务加在该方法。
温馨提示:之所以不在 UserServicveImpl 中执行本地事务,是因为 executeLocalTransaction 中抛出的异常会被 RocketMQ 框架捕捉,及异常无法被 UserServiceImpl 感知,即无法实现其事务的一致性。


接下来展示其核心代码,全部源码已上传到 github 仓库。


仓库地址:https://github.com/dingwpmz/rocketmq-learning


UserServiceImpl 核心实现


image.png


UserServiceImpl 的核心要点如下


  • 首先应该对参数进行校验、业务逻辑进行校验,如果不满足业务条件,会发送一些无效消息到 MQ,虽然不会造成业务异常,但会消耗性能。
  • 发送事务消息,建议对消息设置 Key,Key 的值可以用业务处理流水号(可唯一表示该业务操作)或者核心业务字段(例如订单编号)。
  • 业务入口类可通过事务消息发送状态来判断业务是否失败。


UserRegTransactionListener 核心实现


事务监听器需要实现执行本地事务与事务回查两个接口。


1、实现 executeLocalTransaction


首先需要实现 executeLocalTransaction 方法,执行本地事务,其代码如下图所示:


image.png


其中几个关键点说明如下:


  • 在该方法上添加数据库事务标签。
  • 执行业务逻辑,示例 Demo 只是将用户数据存储到数据库。
  • 如果业务执行失败,可明确告知需要回滚,上层调用方也可根据 ROLLBACK_MESSAGE 进行相应的处理。
  • 如果业务成功,不建议直接返回 COMMIT,而是建议返回 UNKNOW,因为该方法尽管在方法最后一行,但可能发生断电等异常情况,数据库并没有成功。


2、实现 checkLocalTransaction


其次需要实现事务状态回查,用来 RocketMQ 服务端感知事务是否成功,其实现原理如下图所示:


image.png


其实现关键点如下:

  • 如果能明确得知本地事务成功,则返回 COMMIT_MESSAGE
  • 如该不能明确得知本地事务成功,不能返回 ROLLBACK_MESSAGE,而是返回 UNKNOW,等待服务端下一次事务回查(不会立即触发),服务端默认回查 15 次,如果 15 次都得到 UNKNOW,则会回滚该消息。


代码获取


上文只是将事务消息的核心代码加以解读,并重点阐述每个步骤的实现关键点,笔者基于 SpringBoot,尝试结合场景学习 RocketMQ 的使用技巧,其代码上传到了 github 仓库:https://github.com/dingwpmz/rocketmq-learning


image.png


点击跳转到代码仓库


扫码了解更多中间件技术干货和案例实践:

qrcode_for_gh_94efc5c3f960_344.jpg


相关实践学习
RocketMQ一站式入门使用
从源码编译、部署broker、部署namesrv,使用java客户端首发消息等一站式入门RocketMQ。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
相关文章
|
1月前
|
消息中间件 Java API
RocketMQ事务消息, 图文、源码学习探究~
介绍 RocketMQ是阿里巴巴开源的分布式消息中间件,它是一个高性能、低延迟、可靠的消息队列系统,用于在分布式系统中进行异步通信。 从4.3.0版本开始正式支持分布式事务消息~ RocketMq事务消息支持最终一致性:在普通消息基础上,支持二阶段的提交能力。将二阶段提交和本地事务绑定,实现全局提交结果的一致性。 原理、流程 本质上RocketMq的事务能力是基于二阶段提交来实现的 在消息发送上,将二阶段提交与本地事务绑定 本地事务执行成功,则事务消息成功,可以交由Consumer消费 本地事务执行失败,则事务消息失败,Consumer无法消费 但是,RocketMq只能保证本地事务
|
1月前
|
消息中间件 存储 Apache
精华推荐 | 【深入浅出RocketMQ原理及实战】「性能原理挖掘系列」透彻剖析贯穿RocketMQ的事务性消息的底层原理并在分析其实际开发场景
事务消息(Transactional Message)是指应用本地事务和发送消息操作可以被定义到全局事务中,要么同时成功,要么同时失败。RocketMQ的事务消息提供类似 X/Open XA 的分布事务功能,通过事务消息能达到分布式事务的最终一致。
312 2
精华推荐 | 【深入浅出RocketMQ原理及实战】「性能原理挖掘系列」透彻剖析贯穿RocketMQ的事务性消息的底层原理并在分析其实际开发场景
|
7月前
|
PHP 计算机视觉
PHPGrafika 如何实现圆角图片
PHPGrafika 如何实现圆角图片 在网站开发中,圆角图片是非常常见的一种设计元素。使用 PHPGrafika 库可以很方便的实现圆角图片的制作。本文将介绍如何使用 PHPGrafika 库制作圆角图片的方法。
61 0
|
8月前
|
XML 缓存 Java
十一.Spring源码剖析-事务源码之@Transactionl解析
在上一章我们分析了Spring的AOP的源码,本篇文章是对事务的源码分析,我们都知道事务的管理是基于AOP实现的,所以有了上一篇的铺垫这一章会比较简单一点。 事务的源码我会分两章写,一张写Transcational的解析,一张写事务的执行流程。先上一个图,待会儿可以根据这个图来看源码
|
8月前
|
消息中间件 中间件 关系型数据库
【项目实战典型案例】16.消息队列作用和意义
【项目实战典型案例】16.消息队列作用和意义
|
9月前
|
设计模式 存储 开发框架
C++ 插件机制的实现原理、过程、及使用
C++ 插件机制的实现原理、过程、及使用
|
消息中间件 存储 Dubbo
事务消息应用场景、实现原理与项目实战(附全部源码)
从应用场景出发,给出解决方案与实现原理,并提供整套工业级实现源码。
118 0
事务消息应用场景、实现原理与项目实战(附全部源码)
|
消息中间件 存储 缓存
精华推荐 | 【深入浅出RocketMQ原理及实战】「底层原理挖掘系列」透彻剖析贯穿RocketMQ的存储系统的实现原理和持久化机制
精华推荐 | 【深入浅出RocketMQ原理及实战】「底层原理挖掘系列」透彻剖析贯穿RocketMQ的存储系统的实现原理和持久化机制
180 0
精华推荐 | 【深入浅出RocketMQ原理及实战】「底层原理挖掘系列」透彻剖析贯穿RocketMQ的存储系统的实现原理和持久化机制
|
消息中间件 存储 缓存
精华推荐 | 【深入浅出RocketMQ原理及实战】「底层原理挖掘系列」透彻剖析贯穿RocketMQ的Broker服务端自动创建topic的原理分析和问题要点指
精华推荐 | 【深入浅出RocketMQ原理及实战】「底层原理挖掘系列」透彻剖析贯穿RocketMQ的Broker服务端自动创建topic的原理分析和问题要点指
177 0
精华推荐 | 【深入浅出RocketMQ原理及实战】「底层原理挖掘系列」透彻剖析贯穿RocketMQ的Broker服务端自动创建topic的原理分析和问题要点指
|
消息中间件 存储 Java
RocketMQ事务消息原理简析
在项目中,经常遇到这样一个场景,需要保证数据持久化和消息发送要么同时成功,要么同时失败。比如当用户在交易系统下了一个订单,购物车需要消费订单消息清除加购数据、积分系统需要变更用户积分、短信平台需要给买家发送提醒等。利用RocketMQ事务消息特性,可以轻松达到这个目的。本文将从RocketMQ事务消息使用方法说起,探究RocketMQ事务消息实现原理。
200 0

相关产品

  • 云消息队列 MQ
  • 云消息队列 Kafka 版
  • 微服务引擎