RocketMQ事务消息实现分析

简介: 这周RocketMQ发布了4.3.0版本,New Feature中最受关注的一点就是支持了事务消息: 今天花了点时间看了下具体的实现内容,下面是简单的总结。 RocketMQ事务消息概要 通过冯嘉发布的《RocketMQ 4.3正式发布,支持分布式事务》一文可以看到RocketMQ采用了2PC的方案来提交事务消息,同时增加一个补偿逻辑来处理二阶段超时或者失败的消息。

这周RocketMQ发布了4.3.0版本,New Feature中最受关注的一点就是支持了事务消息:

今天花了点时间看了下具体的实现内容,下面是简单的总结。

RocketMQ事务消息概要

通过冯嘉发布的《RocketMQ 4.3正式发布,支持分布式事务》一文可以看到RocketMQ采用了2PC的方案来提交事务消息,同时增加一个补偿逻辑来处理二阶段超时或者失败的消息。

这张图说明了事务消息的大致方案,分为两个逻辑:正常事务消息的发送及提交、事务消息的补偿流程

事务消息发送及提交:

  1. 发送消息(half消息)
  2. 服务端响应消息写入结果
  3. 根据发送结果执行本地事务(如果写入失败,此时half消息对业务不可见,本地逻辑不执行)
  4. 根据本地事务状态执行Commit或者Rollback(Commit操作生成消息索引,消息对消费者可见)

补偿流程:

  1. 对没有Commit/Rollback的事务消息(pending状态的消息),从服务端发起一次“回查”
  2. Producer收到回查消息,检查回查消息对应的本地事务的状态
  3. 根据本地事务状态,重新Commit或者Rollback

补偿阶段用于解决消息Commit或者Rollback发生超时或者失败的情况。

以上RocketMQ事务消息的整体方案,对于了解Notify的同学应该是很熟悉的,下面是之前Notify相关的资料:

整体方案是完全相同的,只是两者的Storage不同。

RocketMQ事务消息设计

一阶段的消息如何对用户不可见

事务消息相对普通消息最大的特点就是一阶段发送的消息对用户是不可见的。

如何做到写入了消息但是对用户不可见?——写入消息数据,但是不创建对应的消息的索引信息。

熟悉RocketMQ的同学应该都清楚,消息在服务端的存储结构如上,每条消息都会有对应的索引信息,Consumer通过索引读取消息。

那么实现一阶段写入的消息不被用户消费(需要在Commit后才能消费),只需要写入Storage Queue,但是不构建Index Queue即可。

RocketMQ中具体实现策略是:写入的如果事务消息,对消息的Topic和Queue等属性进行替换,同时将原来的Topic和Queue信息存储到消息的属性中。

上图即RocketMQ替换事务消息属性的代码实现,替换属性后这条消息被写入到TransactionalMessageUtil.buildHalfTopic()的Queue 0中。

(RocketMQ将事务消息一阶段发送的消息称为Half消息让人费解,采用的2PC的方式,一阶段消息称为Prepare Message或者Pending Message更能体现它的含义)

在完成Storage Queue的写入后,在appendCallback中,普通消息会去构建消息索引,而如果发现是事务消息,则跳过了创建索引的逻辑。

如果让一阶段的消息对用户可见

在完成一阶段写入一条对用户不可见的消息后,二阶段如果是Commit操作,则需要让消息对用户可见;如果是Rollback则需要撤销一阶段的消息。

先说Rollback的情况。对于Rollback,本身一阶段的消息对用户是不可见的,其实不需要真正撤销消息(实际上RocketMQ也无法去真正的删除一条消息,因为是顺序写文件的)。

但是区别于这条消息没有确定状态(Pending状态,事务悬而未决),需要一个操作来标识这条消息的最终状态。

RocketMQ事务消息方案中引入了Op消息的概念,用Op消息标识事务消息是否状态已经确定(Commit或者Rollback)。如果一条事务消息没有对应的Op消息,说明这个事务的状态还无法确定(可能是二阶段失败了)。

引入Op消息后,事务消息无论是Commit或者Rollback都会记录一个Op操作。

Commit相对于Rollback只是在写入Op消息前创建Half消息的索引。

Op消息的存储

RocketMQ将Op消息写入到全局一个特定的Topic中:TransactionalMessageUtil.buildOpTopic()

这个Topic是一个内部的Topic(像Half消息的Topic一样),不会被用户消费。

Op消息的内容为对应的Half消息的存储的Offset,这样通过Op消息能索引到Half消息进行后续的回查操作。

Half消息的索引构建

在执行二阶段的Commit操作时,需要构建出Half消息的索引。

一阶段的Half消息由于是写到一个特殊的Topic,所以二阶段构建索引时需要读取出Half消息,并将Topic和Queue替换成真正的目标的Topic和Queue,之后通过一次普通消息的写入操作来生成一条对用户可见的消息。

所以RocketMQ事务消息二阶段其实是利用了一阶段存储的消息的内容,在二阶段时恢复出一条完整的普通消息,然后走一遍消息写入流程。

如何处理二阶段失败的消息

如果二阶段失败了,比如在Commit操作时出现网络问题导致Commit失败,那么需要通过一定的策略使这条消息最终被Commit。

RocketMQ采用了一种补偿机制,称为“回查”。

Broker端对未确定状态的消息发起回查,将消息发送到对应的Producer端(同一个Group的Producer),由Producer根据消息来检查本地事务的状态,进而执行Commit或者Rollback。

Broker端通过对比Half消息和Op消息进行事务消息的回查并且推进CheckPoint(记录那些事务消息的状态是确定的)。

值得注意的一点是具体实现中,在回查前,系统会执行putBackHalfMsgQueue操作,即将Half消息重新写一遍到Half消息的Queue中。这么做其实是为了能有效的推进上面的CheckPoint。

RocketMQ事务消息设计总结

以上是RocketMQ事务消息实现的示意图:

  • 通过写Half消息的方式来实现一阶段消息对用户不可见
  • 通过Op消息来标记事务消息的状态
  • 通过读取Half消息来生成一条新的Normal消息来完成二阶段Commit之后消息对Consumer可见
  • 通过Op消息来执行回查

优势:

  • Half Queue和Op Queue的数量可控,不会随着Topic的增加而增加
  • 没有外部依赖,实现自包含

缺陷:

  • 每条事务消息至少需要写一条Half消息(异常情况可能会有多条)和Normal,写放大了
  • 所有Half消息都是写到全局预设的一个内部的Topic,这块可能性能会有一些问题(所有Topic的事务消息会往一个Topic上写)
  • 全局Op消息写一个Topic,回查时间可能会有相互影响

如果本文对您有帮助,点一下右下角的“推荐”
相关实践学习
消息队列RocketMQ版:基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能,涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
目录
相关文章
|
4月前
|
消息中间件 监控 数据挖掘
基于RabbitMQ与Apache Flink构建实时分析系统
【8月更文第28天】本文将介绍如何利用RabbitMQ作为数据源,结合Apache Flink进行实时数据分析。我们将构建一个简单的实时分析系统,该系统能够接收来自不同来源的数据,对数据进行实时处理,并将结果输出到另一个队列或存储系统中。
294 2
|
2月前
|
消息中间件 Java 数据库
新版 Seata 集成 RocketMQ事务消息,越来越 牛X 了!阿里的 Seata , yyds !
这里 借助 Seata 集成 RocketMQ 事务消息的 新功能,介绍一下一个新遇到的面试题:如果如何实现 **强弱一致性 结合**的分布式事务?
新版 Seata 集成 RocketMQ事务消息,越来越 牛X 了!阿里的 Seata , yyds !
|
2月前
|
消息中间件 监控 供应链
深度剖析 RocketMQ 事务消息!
本文深入探讨了 RocketMQ 的事务消息原理及其应用场景。通过详细的源码分析,阐述了事务消息的基本流程,包括准备阶段、提交阶段及补偿机制。文章还提供了示例代码,帮助读者更好地理解整个过程。此外,还讨论了事务消息的优缺点、适用场景及注意事项,如确保本地事务的幂等性、合理设置超时时间等。尽管事务消息增加了系统复杂性,但在需要保证消息一致性的场景中,它仍是一种高效的解决方案。
150 2
|
7月前
|
消息中间件 存储 RocketMQ
RocketMQ源码分析之事务消息实现原理下篇-消息服务器Broker提交回滚事务实现原理
RocketMQ源码分析之事务消息实现原理下篇-消息服务器Broker提交回滚事务实现原理
|
4月前
|
消息中间件 存储 数据中心
RocketMQ的长轮询(Long Polling)实现分析
文章深入分析了RocketMQ的长轮询实现机制,长轮询结合了推送(push)和拉取(pull)两种消息消费模式的优点,通过客户端和服务端的配合,确保了消息的实时性同时将主动权保留在客户端。文中首先解释了长轮询的基本概念和实现步骤,然后通过一个简单的实例模拟了长轮询的过程,最后详细介绍了RocketMQ中DefaultMQPushConsumer的长轮询实现方式,包括PullMessage服务、PullMessageProcessor服务和PullCallback回调的工作原理。
132 1
|
4月前
|
消息中间件 存储 缓存
RocketMQ发送消息原理(含事务消息)
本文深入探讨了RocketMQ发送消息的原理,包括生产者端的发送流程、Broker端接收和处理消息的流程,以及事务消息的特殊处理机制,提供了对RocketMQ消息发送机制全面的理解。
RocketMQ发送消息原理(含事务消息)
|
4月前
|
消息中间件 Arthas Java
RocketMQ—一次连接namesvr失败的案例分析
项目组在使用RocketMQ时遇到Consumer连接Name Server失败的问题,异常显示连接特定地址失败。通过Arthas工具逐步分析代码执行路径,定位到创建Channel返回空值导致异常。进一步跟踪发现,问题源于Netty组件在初始化`ByteBufAllocator`时出现错误。分析依赖后确认存在Netty版本冲突。解决方法为排除冲突的Netty包,仅保留兼容版本。
308 0
RocketMQ—一次连接namesvr失败的案例分析
|
5月前
|
消息中间件 Java 测试技术
【RocketMQ系列八】SpringBoot集成RocketMQ-实现普通消息和事务消息
【RocketMQ系列八】SpringBoot集成RocketMQ-实现普通消息和事务消息
396 1
|
5月前
|
消息中间件 调度 RocketMQ
【RocketMQ系列六】RocketMQ事务消息
【RocketMQ系列六】RocketMQ事务消息
1008 1
|
6月前
|
数据采集 监控 物联网
MQTT协议在智能制造中的应用案例与效益分析
【6月更文挑战第8天】MQTT协议在智能制造中的应用案例与效益分析
199 1