一文让你事务消息应用场景、实现原理与项目实战(附全部源码)
共 2957字，需浏览 6分钟
·
2021-03-11 12:50
大家好，我是跃哥。平时开发的过程中，为了增强程序的性能，小伙伴们基本上都会使用消息中间件。那今天就根据业务场景，来逐步分析如何使用RocketMQ事务消息，并懂原理。冲鸭
1、活动中心场景介绍
在电商系统上线初期，往往会进行一些“拉新”活动，例如活动部门提出新用户注册送积分、送优惠券活动。
基于分布式、微服务的设计理念，通常的架构设计（子系统交互）如下图所示：
其核心系统介绍如下：
账户中心
提供用户登录、用户注册等服务，一个新用户注册时，向MQ服务器中的USER_REGISTER主题发送一条消息，主流程结束，与送积分，送优惠券等过程解耦。优惠券（券系统）
提供发放优惠券、使用优惠券等与券相关的基础服务。积分中心
提供积分相关的服务，例如积分赠送、积分消费、积分查询等基础服务。送积分服务（消费者）
订阅MQ，按照规则决定是否需要赠送积分，如果需要则调用积分相关的基础接口，完成积分的发放。送优惠券（消费者）
订阅MQ，按照规则决定是否需要赠送优惠券，如果需要则调用券系统相关的基础接口，完成优惠券的发放。
上面的架构设计非常优雅，但并不是无懈可击，读者们肯定会想到如果新用户注册成功，但消息发送到MQ失败，或者消息成功发送到MQ，但发送完MQ后系统出现异常导致用户注册失败又该如何呢？
上面的问题其实就是典型的分布式事务问题：即如何保证用户注册(数据库操作)与MQ消息发送这两个分布式操作的一致性。
RocketMQ事务消息闪亮登场。
2、事务消息实现原理
一言以蔽之：RocketMQ事务消息要解决的问题是消息发送与业务的一致性，其解决思路：二阶段提交与事务状态回查，其具体实现流程如下图所示：
其核心设计理念：
应用程序开启一个数据库事务，进行数据库操作，并且在事务中发送一条PREPARE消息，PREPARE消息发送成功后通知应用程序记录本地事务状态，然后提交本地事务。
RocketMQ在收到类型为PREPARE的消息时，首先备份消息的原主题与原消息消费队列，然后将消息存储在主题为RMQ_SYS_TRANS_HALF_TOPIC的消息队列中，故PREPARE的消息是不会被客户端消费的。
Broker消息服务器开启一个定时任务处理RMQ_SYS_TRANS_HALF_TOPIC中的消息，会每隔指定时间向消息发送者发起事务状态查询请求 ,询问消息发送者客户端本地事务是否成功，然后根据回查状态决定是提交还是回滚，即对处于PREPARE状态进行提交或回滚操作。
发送者如果明确得知事务成功，则可以返回COMMIT，服务端会提交该条消息，具体操作是恢复原消息的主题与队列，重新发送到Broker，消费端感知后消费。
发送者如果无法明确得知事务状态，则返回UNOWN，此时服务端会等待一定时间后再次向发送者询问，默认询问15次。
发送者如果非常明确得知事务失败，则可以返回ROLLBACK。
在具体实践中，消息发送者在无法获取事务状态时不要武断的返回ROLLBACK，而是要返回UNOWN，让服务端定时重试回查，说明如下：
在将PREPARE消息发送到Broker后，服务端发起事务查询时本地事务可能还未提交，为了避免无效的事务回查机制，RocketMQ通常至少在收到PREPARE消息6s后才会发起第一次事务回查，可通过 transactionTimeOut 配置。故客户端在实现事务回查时无法证明事务状态时不应该返回ROLLBACK，而是返回UNOWN。
3、事务消息实战
光说不练假把式，接下来以一个新用户注册送优惠券的场景来详细介绍如何使用事务消息。
项目模块职责说明如下：
事务消息的核心代码组装在transaction-service，其核心类图如下：
其中核心要点如下：
UserServiceImpl
Dubbo接口业务实现类，类似MVC的控制层，在这里做一些参数验证，但不执行具体的业务逻辑，只是发送一条事务消息到MQ。UserRegTransactionListener
事务监听器，在 executeLocalTransaction 方法中执行业务逻辑，数据库本地事务加在该方法。
温馨提示：之所以不在UserServicveImpl中执行本地事务，是因为 executeLocalTransaction 中抛出的异常会被RocketMQ框架捕捉，及异常无法被UserServiceImpl感知，即无法实现其事务的一致性。
接下来展示其核心代码，全部源码已上传到github仓库。
仓库地址：https://github.com/dingwpmz/rocketmq-learning
3.1 UserServiceImpl 核心实现
UserServiceImpl 的核心要点如下：
首先应该对参数进行校验、业务逻辑进行校验，如果不满足业务条件，会发送一些无效消息到MQ，虽然不会造成业务异常，但会消耗性能
发送事务消息，建议对消息设置Key,Key的值可以用业务处理流水号(可唯一表示该业务操作)或者核心业务字段(例如订单编号)
业务入口类可通过事务消息发送状态来判断业务是否失败。
3.2 UserRegTransactionListener 核心实现
事务监听器需要实现执行本地事务与事务回查两个接口。
3.2.1 实现 executeLocalTransaction
首先需要实现 executeLocalTransaction 方法，执行本地事务，其代码如下图所示：
其中几个关键点说明如下：
在该方法上添加数据库事务标签。
执行业务逻辑，示例Demo只是将用户数据存储到数据库。
如果业务执行失败，可明确告知需要回滚，上层调用方也可根据ROLLBACK_MESSAGE进行相应的处理。
如果业务成功，不建议直接返回COMMIT，而是建议返回UNKNOW,因为该方法尽管在方法最后一行，但可能发生断电等异常情况，数据库并没有成功。
3.2.2 实现 checkLocalTransaction
其次需要实现事务状态回查，用来RocketMQ服务端感知事务是否成功，其实现原理如下图所示：
其实现关键点如下：
如果能明确得知本地事务成功，则返回COMMIT_MESSAGE
如该不能明确得知本地事务成功，不能返回ROLLBACK_MESSAGE,而是返回UNKNOW，等待服务端下一次事务回查(不会立即触发)，服务端默认回查15次，如果15次都得到UNKNOW，则会回滚该消息。
3.3 代码获取
上文只是将事务消息的核心代码加以解读，并重点阐述每个步骤的实现关键点，笔者基于SpringBoot，尝试结合场景学习RocketMQ的使用技巧，其代码上传到了github仓库。
https://github.com/dingwpmz/rocketmq-learning
点击「阅读原文」可跳转到代码仓库。
