聊聊 RocketMQ 事务消息？

事务消息是 RocketMQ 的高级特性之一 。这篇文章笔者会从应用场景、功能原理、实战例子三个模块慢慢为你揭开事务消息的神秘面纱。1 应用场景举一个电商场景的例子用户购物车结算时系统会创建支付订单。用户支付成功后支付订单的状态会由未支付修改为支付成功然后系统给用户增加积分。通常我们会使用普通消费方案该方案能够发挥 MQ 的优势异步和解耦, 同时架构设计非常简单。用户购物车结算时系统创建支付订单支付成功后更新订单的状态从未支付修改为支付成功发送一条普通消息到消息队列服务端;积分服务消费消息添加积分记录。但该方案有个非常直观的缺点容易出现不一致的现象。假如先发送消息后修改订单状态消息发送成功订单没有执行成功需要回滚整个事务订单数据事务回滚积分服务消费时需要先反查事务状态若事务提交才能插入积分记录。假如先修改订单状态后发送消息订单状态修改成功但消息发送失败需要补偿操作才能保持最终一致。假如先修改订单后发送消息订单状态修改成功但消息发送超时此时无法判断需要回滚订单还是提交订单变更。我们看到为了完善普通消费方案业务层还需要做到两点补偿机制和提供事务状态查询接口。要做到这两点难不难呢不难但是业务层代码会比较混乱更优的方案还是得从中间件层面解决。2 功能原理RocketMQ 事务消息是支持在分布式场景下保障消息生产和本地事务的最终一致性。交互流程如下图所示1、生产者将消息发送至 Broker 。2、Broker 将消息持久化成功之后向生产者返回 Ack 确认消息已经发送成功此时消息被标记为暂不能投递这种状态下的消息即为半事务消息。3、生产者开始执行本地事务逻辑。4、生产者根据本地事务执行结果向服务端提交二次确认结果 Commit 或是 Rollback Broker 收到确认结果后处理逻辑如下二次确认结果为 Commit Broker 将半事务消息标记为可投递并投递给消费者。二次确认结果为 Rollback Broker 将回滚事务不会将半事务消息投递给消费者。5、在断网或者是生产者应用重启的特殊情况下若 Broker 未收到发送者提交的二次确认结果或 Broker 收到的二次确认结果为 Unknown 未知状态经过固定时间后服务端将对消息生产者即生产者集群中任一生产者实例发起消息回查。生产者收到消息回查后需要检查对应消息的本地事务执行的最终结果。生产者根据检查到的本地事务的最终状态再次提交二次确认服务端仍按照步骤4对半事务消息进行处理。笔者认为事务消息的精髓在于本地事务执行成功消费者才能消费事务消息消息回查本身就是补偿机制的实现事务生产者需提供了事务状态查询接口。3 实战例子为了便于大家理解事务消息 笔者新建一个工程用于模拟支付订单创建、支付成功、赠送积分的流程。首先我们创建一个真实的订单主题order-topic。然后在数据库中创建三张表订单表、事务日志表、积分表。最后我们创建一个 Demo 工程生产者模块用于创建支付订单、修改支付订单成功消费者模块用于新增积分记录。接下来我们展示事务消息的实现流程。1、创建支付订单调用订单生产者服务创建订单接口 在 t_order 表中插入一条支付订单记录。2、调用生产者服务修改订单状态接口接口的逻辑就是执行事务生产者的sendMessageInTransaction方法。生产者端需要配置事务生产者和事务监听器。发送事务消息的方法内部包含三个步骤 事务生产者首先发送半事务消息发送成功后生产者才开始执行本地事务逻辑。事务监听器实现了两个功能执行本地事务和供 Broker 回查事务状态。执行本地事务的逻辑内部就是执行orderService.updateOrder方法。方法执行成功则返回LocalTransactionState.COMMIT_MESSAGE, 若执行失败则返回LocalTransactionState.ROLLBACK_MESSAGE。需要注意的是orderService.updateOrder方法添加了事务注解并将修改订单状态和插入事务日志表放进一个事务内避免订单状态和事务日志表的数据不一致。最后生产者根据本地事务执行结果向 Broker提交二次确认结果。Broker 收到生产者确认结果后处理逻辑如下二次确认结果为 Commit Broker 将半事务消息标记为可投递并投递给消费者。二次确认结果为 Rollback Broker 将回滚事务不会将半事务消息投递给消费者。3、积分消费者消费消息添加积分记录当 Broker 将半事务消息标记为可投递时积分消费者就可以开始消费主题 order-topic 的消息了。积分消费者服务我们定义了消费者组名以及订阅主题和消费监听器。在消费监听器逻辑里幂等非常重要。当收到订单信息后首先判断该订单是否有积分记录若没有记录才插入积分记录。而且我们在创建积分表时订单编号也是唯一键数据库中也必然不会存在相同订单的多条积分记录。4 总结RocketMQ 事务消息是支持在分布式场景下保障消息生产和本地事务的最终一致性。编写一个实战例子并不复杂但使用事务消息时需要注意如下三点1、事务生产者和消费者共同协作才能保证业务数据的最终一致性2、事务生产者需要实现事务监听器并且保存事务的执行结果比如事务日志表 3、消费者要保证幂等。消费失败时通过重试、告警人工介入等手段保证消费结果正确。