当前位置: 首页 > news >正文

RocketMQ 事务消息发送

目录

事务消息介绍

应用场景

功能原理

使用限制

使用示例

使用建议​


事务消息介绍

在一些对数据一致性有强需求的场景,可以用 RocketMQ 事务消息来解决,从而保证上下游数据的一致性。

应用场景

分布式事务的诉求

分布式系统调用的特点为一个核心业务逻辑的执行,同时需要调用多个下游业务进行处理。因此,如何保证核心业务和多个下游业务的执行结果完全一致,是分布式事务需要解决的主要问题。

以电商交易场景为例,用户支付订单这一核心操作的同时会涉及到下游物流发货、积分变更、购物车状态清空等多个子系统的变更。当前业务的处理分支包括:

  • 主分支订单系统状态更新:由未支付变更为支付成功。

  • 物流系统状态新增:新增待发货物流记录,创建订单物流记录。

  • 积分系统状态变更:变更用户积分,更新用户积分表。

  • 购物车系统状态变更:清空购物车,更新用户购物车记录。

传统XA事务方案:性能不足

为了保证上述四个分支的执行结果一致性,典型方案是基于XA协议的分布式事务系统来实现。将四个调用分支封装成包含四个独立事务分支的大事务。基于XA分布式事务的方案可以满足业务处理结果的正确性,但最大的缺点是多分支环境下资源锁定范围大,并发度低,随着下游分支的增加,系统性能会越来越差。

基于普通消息方案:一致性保障困难

将上述基于XA事务的方案进行简化,将订单系统变更作为本地事务,剩下的系统变更作为普通消息的下游来执行,事务分支简化成普通消息+订单表事务,充分利用消息异步化的能力缩短链路,提高并发度。

该方案中消息下游分支和订单系统变更的主分支很容易出现不一致的现象,例如:

  • 消息发送成功,订单没有执行成功,需要回滚整个事务。

  • 订单执行成功,消息没有发送成功,需要额外补偿才能发现不一致。

  • 消息发送超时未知,此时无法判断需要回滚订单还是提交订单变更。

基于RocketMQ分布式事务消息:支持最终一致性

上述普通消息方案中,普通消息和订单事务无法保证一致的原因,本质上是由于普通消息无法像单机数据库事务一样,具备提交、回滚和统一协调的能力。

而基于Apache RocketMQ实现的分布式事务消息功能,在普通消息基础上,支持二阶段的提交能力。将二阶段提交和本地事务绑定,实现全局提交结果的一致性。

Apache RocketMQ事务消息的方案,具备高性能、可扩展、业务开发简单的优势。具体事务消息的原理和流程,请参见下文的功能原理。

功能原理

什么是事务消息

事务消息是 RocketMQ 提供的一种高级消息类型,支持在分布式场景下保障消息生产和本地事务的最终一致性。

事务消息处理流程

事务消息交互流程如下图所示。

  1. 生产者将消息发送至RocketMQ服务端。

  2. RocketMQ服务端将消息持久化成功之后,向生产者返回Ack确认消息已经发送成功,此时消息被标记为"暂不能投递",这种状态下的消息即为半事务消息。

  3. 生产者开始执行本地事务逻辑。

  4. 生产者根据本地事务执行结果向服务端提交二次确认结果(Commit或是Rollback),服务端收到确认结果后处理逻辑如下:

    • 二次确认结果为Commit:服务端将半事务消息标记为可投递,并投递给消费者。

    • 二次确认结果为Rollback:服务端将回滚事务,不会将半事务消息投递给消费者。

  5. 在断网或者是生产者应用重启的特殊情况下,若服务端未收到发送者提交的二次确认结果,或服务端收到的二次确认结果为Unknown未知状态,经过固定时间后,服务端将对消息生产者即生产者集群中任一生产者实例发起消息回查。 说明 服务端回查的间隔时间和最大回查次数,请参见参数限制。

  6. 生产者收到消息回查后,需要检查对应消息的本地事务执行的最终结果。

  7. 生产者根据检查到的本地事务的最终状态再次提交二次确认,服务端仍按照步骤4对半事务消息进行处理。

 

  • 初始化:半事务消息被生产者构建并完成初始化,待发送到服务端的状态。

  • 事务待提交:半事务消息被发送到服务端,和普通消息不同,并不会直接被服务端持久化,而是会被单独存储到事务存储系统中,等待第二阶段本地事务返回执行结果后再提交。此时消息对下游消费者不可见。

  • 消息回滚:第二阶段如果事务执行结果明确为回滚,服务端会将半事务消息回滚,该事务消息流程终止。

  • 提交待消费:第二阶段如果事务执行结果明确为提交,服务端会将半事务消息重新存储到普通存储系统中,此时消息对下游消费者可见,等待被消费者获取并消费。

  • 消费中:消息被消费者获取,并按照消费者本地的业务逻辑进行处理的过程。 此时服务端会等待消费者完成消费并提交消费结果,如果一定时间后没有收到消费者的响应,Apache RocketMQ会对消息进行重试处理。具体信息,请参见消费重试。

  • 消费提交:消费者完成消费处理,并向服务端提交消费结果,服务端标记当前消息已经被处理(包括消费成功和失败)。 Apache RocketMQ默认支持保留所有消息,此时消息数据并不会立即被删除,只是逻辑标记已消费。消息在保存时间到期或存储空间不足被删除前,消费者仍然可以回溯消息重新消费。

  • 消息删除:Apache RocketMQ按照消息保存机制滚动清理最早的消息数据,将消息从物理文件中删除。更多信息,请参见消息存储和清理机制。

使用限制

消息类型一致性

事务消息仅支持在 MessageType 为 Transaction 的主题内使用,即事务消息只能发送至类型为事务消息的主题中,发送的消息的类型必须和主题的类型一致。

消费事务性

RocketMQ 事务消息保证本地主分支事务和下游消息发送事务的一致性,但不保证消息消费结果和上游事务的一致性。因此需要下游业务分支自行保证消息正确处理,建议消费端做好消费重试,如果有短暂失败可以利用重试机制保证最终处理成功。

中间状态可见性

RocketMQ 事务消息为最终一致性,即在消息提交到下游消费端处理完成之前,下游分支和上游事务之间的状态会不一致。因此,事务消息仅适合接受异步执行的事务场景。

事务超时机制

RocketMQ 事务消息的命周期存在超时机制,即半事务消息被生产者发送服务端后,如果在指定时间内服务端无法确认提交或者回滚状态,则消息默认会被回滚。事务超时时间,请参见参数限制。

使用示例

创建主题

RocketMQ 5.0版本下创建主题操作,推荐使用mqadmin工具,需要注意的是,对于消息类型需要通过属性参数添加。示例如下:

sh mqadmin updateTopic -n <nameserver_address> -t <topic_name> -c <cluster_name> -a +message.type=TRANSACTION

发送消息

事务消息相比普通消息发送时需要修改以下几点:

  • 发送事务消息前,需要开启事务并关联本地的事务执行。

  • 为保证事务一致性,在构建生产者时,必须设置事务检查器和预绑定事务消息发送的主题列表,客户端内置的事务检查器会对绑定的事务主题做异常状态恢复。

创建事务主题

NORMAL类型Topic不支持TRANSACTION类型消息,生产消息会报错。

./bin/mqadmin updatetopic -n localhost:9876 -t TestTopic -c DefaultCluster -a +message.type=TRANSACTION

  • -c 集群名称
  • -t Topic名称
  • -n nameserver地址
  • -a 额外属性,本例给主题添加了message.typeTRANSACTION的属性用来支持事务消息

以Java语言为例,使用事务消息示例参考如下:

    //演示demo,模拟订单表查询服务,用来确认订单事务是否提交成功。private static boolean checkOrderById(String orderId) {return true;}//演示demo,模拟本地事务的执行结果。private static boolean doLocalTransaction() {return true;}public static void main(String[] args) throws ClientException {ClientServiceProvider provider = new ClientServiceProvider();MessageBuilder messageBuilder = new MessageBuilderImpl();//构造事务生产者:事务消息需要生产者构建一个事务检查器,用于检查确认异常半事务的中间状态。Producer producer = provider.newProducerBuilder().setTransactionChecker(messageView -> {/*** 事务检查器一般是根据业务的ID去检查本地事务是否正确提交还是回滚,此处以订单ID属性为例。* 在订单表找到了这个订单,说明本地事务插入订单的操作已经正确提交;如果订单表没有订单,说明本地事务已经回滚。*/final String orderId = messageView.getProperties().get("OrderId");if (Strings.isNullOrEmpty(orderId)) {// 错误的消息,直接返回Rollback。return TransactionResolution.ROLLBACK;}return checkOrderById(orderId) ? TransactionResolution.COMMIT : TransactionResolution.ROLLBACK;}).build();//开启事务分支。final Transaction transaction;try {transaction = producer.beginTransaction();} catch (ClientException e) {e.printStackTrace();//事务分支开启失败,直接退出。return;}Message message = messageBuilder.setTopic("topic")//设置消息索引键,可根据关键字精确查找某条消息。.setKeys("messageKey")//设置消息Tag,用于消费端根据指定Tag过滤消息。.setTag("messageTag")//一般事务消息都会设置一个本地事务关联的唯一ID,用来做本地事务回查的校验。.addProperty("OrderId", "xxx")//消息体。.setBody("messageBody".getBytes()).build();//发送半事务消息final SendReceipt sendReceipt;try {sendReceipt = producer.send(message, transaction);} catch (ClientException e) {//半事务消息发送失败,事务可以直接退出并回滚。return;}/*** 执行本地事务,并确定本地事务结果。* 1. 如果本地事务提交成功,则提交消息事务。* 2. 如果本地事务提交失败,则回滚消息事务。* 3. 如果本地事务未知异常,则不处理,等待事务消息回查。**/boolean localTransactionOk = doLocalTransaction();if (localTransactionOk) {try {transaction.commit();} catch (ClientException e) {// 业务可以自身对实时性的要求选择是否重试,如果放弃重试,可以依赖事务消息回查机制进行事务状态的提交。e.printStackTrace();}} else {try {transaction.rollback();} catch (ClientException e) {// 建议记录异常信息,回滚异常时可以无需重试,依赖事务消息回查机制进行事务状态的提交。e.printStackTrace();}}}

使用建议​

避免大量未决事务导致超时

Apache RocketMQ支持在事务提交阶段异常的情况下发起事务回查,保证事务一致性。但生产者应该尽量避免本地事务返回未知结果。大量的事务检查会导致系统性能受损,容易导致事务处理延迟。

正确处理"进行中"的事务

消息回查时,对于正在进行中的事务不要返回Rollback或Commit结果,应继续保持Unknown的状态。 一般出现消息回查时事务正在处理的原因为:事务执行较慢,消息回查太快。解决方案如下:

  • 将第一次事务回查时间设置较大一些,但可能导致依赖回查的事务提交延迟较大。

  • 程序能正确识别正在进行中的事务。

相关文章:

RocketMQ 事务消息发送

目录 事务消息介绍 应用场景 功能原理 使用限制 使用示例 使用建议​ 事务消息介绍 在一些对数据一致性有强需求的场景&#xff0c;可以用 RocketMQ 事务消息来解决&#xff0c;从而保证上下游数据的一致性。 应用场景 分布式事务的诉求 分布式系统调用的特点为一个核…...

后端-POST请求中只需要两个参数,后端不想创建对象时

问题&#xff1a; 在前后端对接中&#xff0c;很多前端的规范是POST&#xff0c;参数放body里面&#xff0c;媒体类型是json&#xff0c;后端就需要用RequestBody去接收&#xff0c;但是后端只用接收两个对象&#xff0c;这时候后端不想创建对象&#xff0c;使用RequestParm&a…...

UG\NX二次开发 通过点云生成曲面 UF_MODL_create_surf_from_cloud

文章作者:里海 来源网站:《里海NX二次开发3000例专栏》 感谢粉丝订阅 感谢 Rlgun 订阅本专栏,非常感谢。 简介 有网友想做一个通过点云生成曲面的程序,我们也试一下 效果 代码 #include "me.hpp" /*HEAD CREATE_SURF_FROM_CLOUD CCC UFUN */...

Linux常用指令(二)

目录 一、 删除空目录&#xff08;rmdir&#xff09; 二、ln 硬链接与软链接 三、新建空文件或更新文件的时间戳&#xff08;touch&#xff09; 四、比较文件内容的差异&#xff08;diff&#xff09; 五、显示当前时间或设置系统时间&#xff08;date&#xff09; 六、显…...

【HUAWEI】单臂路由

目录 ​ &#x1f96e;写在前面 &#x1f96e;2.1、拓扑图 &#x1f96e;2.2、操作思路 &#x1f96e;2.3、配置操作 &#x1f363;2.3.1、LSW4配置 &#x1f363;2.3.2、R2配置 &#x1f363;2.3.3、测试网络 &#x1f990;博客主页&#xff1a;大虾好吃吗的博客 &…...

安全学习_开发相关_Java第三方组件Log4jFastJSON及相关安全问题简介

文章目录 JNDI&#xff1a;(见图) Java-三方组件-Log4J&JNDILog4J&#xff1a;Log4j-组件安全复现使用Log4j Java-三方组件-FastJsonFastJson&#xff1a;Fastjson-组件安全复现对象转Json(带类型)Json转对象Fastjson漏洞复现&#xff08;大佬文章 JNDI&#xff1a;(见图) …...

零代码编程:用ChatGPT批量自动下载archive.org上的音频书

http://archive.org 是一个神奇的网站&#xff0c;可以下载各种古旧的软件、书籍、音频、视频&#xff0c;还可以搜索各个网站的历史网页。 比如说&#xff0c;一些儿童故事音频就可以在http://archive.org下载到&#xff0c;可以用来做英语听力启蒙用。 举个例子&#xff0c…...

力扣用队列实现栈

自己写的栈&#xff0c;再让其他函数去调用自己写的栈 typedef int QDataType; typedef struct QueueNode {struct QueueNode* next;//单链表QDataType data;//放数据 }QNode;typedef struct Queue {QNode* phead;//头节点QNode* ptail;//尾节点QDataType size; //统计有多少节…...

一朵华为云,如何做好百模千态?

点击关注 文丨刘雨琦、郝鑫 2005年华为提出网络时代的“All IP”&#xff0c;2011年提出数字化时代的“All Cloud”&#xff0c;2023年提出智能时代的“All Intelligence”。 截至目前&#xff0c;华为的战略升级经历了三个阶段。 步入智能化&#xff0c;需要迎接的困难依然…...

华为云云耀云服务器L实例评测 | 实例使用教学之软件安装:华为云云耀云服务器环境下安装 Docker

华为云云耀云服务器L实例评测 &#xff5c; 实例使用教学之软件安装&#xff1a;华为云云耀云服务器环境下安装 Docker 介绍华为云云耀云服务器 华为云云耀云服务器 &#xff08;目前已经全新升级为 华为云云耀云服务器L实例&#xff09; 华为云云耀云服务器是什么华为云云耀云…...

小程序编译器性能优化之路

作者 | 马可 导读 小程序编译器是百度开发者工具中的编译构建模块&#xff0c;用来将小程序代码转换成运行时代码。旧版编译器由于业务发展&#xff0c;存在编译慢、内存占用高的问题&#xff0c;我们对编译器做了一次大规模的重构&#xff0c;采用自研架构&#xff0c;做了多线…...

FFmpeg 命令:从入门到精通 | ffmpeg 命令分类查询

FFmpeg 命令&#xff1a;从入门到精通 | ffmpeg 命令分类查询 FFmpeg 命令&#xff1a;从入门到精通 | ffmpeg 命令分类查询ffmpeg -versionffmpeg -buildconfffmpeg -formatsffmpeg -muxersffmpeg -demuxersffmpeg -codecsffmpeg -decodersffmpeg -encodersffmpeg -bsfsffmpeg…...

Linux学习记录——삼십일 socket编程---TCP套接字

文章目录 TCP套接字简单通信1、服务端1、基本框架2、获取连接 2、客户端3、多进程4、多线程5、线程池6、简单的日志系统7、守护进程8、其它 TCP套接字简单通信 本篇gitee 学习完udp套接字通信后&#xff0c;再来看TCP套接字。 四个文件tcp_server.hpp&#xff0c; tcp_serve…...

【学习笔记】深度学习分布式系统

深度学习分布式系统 前言1. 数据并行&#xff1a;参数服务器2. 流水线并行&#xff1a;GPipe3. 张量并行&#xff1a;Megatron LM4. 切片并行&#xff1a;ZeRO5. 异步分布式&#xff1a;PATHWAYS总结参考链接 前言 最近跟着李沐老师的视频学习了深度学习分布式系统的发展。这里…...

【数据结构】树、二叉树的概念和二叉树的顺序结构及实现

目录 前言&#xff1a;一、树的概念及结构1.树的概念2.树的相关概念3.树的存储4.树在实际中的运用 二、二叉树概念及结构1.概念2.特殊的二叉树&#xff08;1&#xff09;满二叉树&#xff08;2&#xff09;完全二叉树 3.二叉树的性质4.二叉树的存储(1)顺序存储(2)链式存储 三、…...

rust学习-string

介绍 A UTF-8–encoded, growable string(可增长字符串). 拥有string内容的所有权 A String is made up of three components: a pointer to some bytes, a length, and a capacity. The length is the number of bytes currently stored in the buffer pub fn as_bytes(&…...

No167.精选前端面试题,享受每天的挑战和学习

🤍 前端开发工程师(主业)、技术博主(副业)、已过CET6 🍨 阿珊和她的猫_CSDN个人主页 🕠 牛客高级专题作者、在牛客打造高质量专栏《前端面试必备》 🍚 蓝桥云课签约作者、已在蓝桥云课上架的前后端实战课程《Vue.js 和 Egg.js 开发企业级健康管理项目》、《带你从入…...

【python】pycharm导入anaconda环境

参考 Pycharm导入anaconda环境的教程图解 - 知乎 (zhihu.com)...

【数据结构】逻辑结构与物理结构

&#x1f984;个人主页:修修修也 &#x1f38f;所属专栏:数据结构 ⚙️操作环境:Visual Studio 2022 目录 &#x1f333;逻辑结构 1.集合结构 2.线性结构 3.树形结构 4.图形结构或网状结构 &#x1f333;物理结构 1.顺序存储结构 2.链式存储结构 结语 根据视点的不同,我…...

HTML5高级部分

目录 一、拖拽API1.1 拖拽元素1.2 监听事件1.3 dataTransfer传递数据 二、媒体API2.1 常用监听事件2.2 常用API 三、画布API3.1 canvas 标签3.2 创建canvas对象3.3 常用API 四、地理API4.1 方法 一、拖拽API 1.1 拖拽元素 页面中设置了draggable"true"的元素可以进…...

通过Wrangler CLI在worker中创建数据库和表

官方使用文档&#xff1a;Getting started Cloudflare D1 docs 创建数据库 在命令行中执行完成之后&#xff0c;会在本地和远程创建数据库&#xff1a; npx wranglerlatest d1 create prod-d1-tutorial 在cf中就可以看到数据库&#xff1a; 现在&#xff0c;您的Cloudfla…...

质量体系的重要

质量体系是为确保产品、服务或过程质量满足规定要求&#xff0c;由相互关联的要素构成的有机整体。其核心内容可归纳为以下五个方面&#xff1a; &#x1f3db;️ 一、组织架构与职责 质量体系明确组织内各部门、岗位的职责与权限&#xff0c;形成层级清晰的管理网络&#xf…...

ESP32 I2S音频总线学习笔记(四): INMP441采集音频并实时播放

简介 前面两期文章我们介绍了I2S的读取和写入&#xff0c;一个是通过INMP441麦克风模块采集音频&#xff0c;一个是通过PCM5102A模块播放音频&#xff0c;那如果我们将两者结合起来&#xff0c;将麦克风采集到的音频通过PCM5102A播放&#xff0c;是不是就可以做一个扩音器了呢…...

什么是Ansible Jinja2

理解 Ansible Jinja2 模板 Ansible 是一款功能强大的开源自动化工具&#xff0c;可让您无缝地管理和配置系统。Ansible 的一大亮点是它使用 Jinja2 模板&#xff0c;允许您根据变量数据动态生成文件、配置设置和脚本。本文将向您介绍 Ansible 中的 Jinja2 模板&#xff0c;并通…...

今日学习:Spring线程池|并发修改异常|链路丢失|登录续期|VIP过期策略|数值类缓存

文章目录 优雅版线程池ThreadPoolTaskExecutor和ThreadPoolTaskExecutor的装饰器并发修改异常并发修改异常简介实现机制设计原因及意义 使用线程池造成的链路丢失问题线程池导致的链路丢失问题发生原因 常见解决方法更好的解决方法设计精妙之处 登录续期登录续期常见实现方式特…...

Web 架构之 CDN 加速原理与落地实践

文章目录 一、思维导图二、正文内容&#xff08;一&#xff09;CDN 基础概念1. 定义2. 组成部分 &#xff08;二&#xff09;CDN 加速原理1. 请求路由2. 内容缓存3. 内容更新 &#xff08;三&#xff09;CDN 落地实践1. 选择 CDN 服务商2. 配置 CDN3. 集成到 Web 架构 &#xf…...

站群服务器的应用场景都有哪些?

站群服务器主要是为了多个网站的托管和管理所设计的&#xff0c;可以通过集中管理和高效资源的分配&#xff0c;来支持多个独立的网站同时运行&#xff0c;让每一个网站都可以分配到独立的IP地址&#xff0c;避免出现IP关联的风险&#xff0c;用户还可以通过控制面板进行管理功…...

Scrapy-Redis分布式爬虫架构的可扩展性与容错性增强:基于微服务与容器化的解决方案

在大数据时代&#xff0c;海量数据的采集与处理成为企业和研究机构获取信息的关键环节。Scrapy-Redis作为一种经典的分布式爬虫架构&#xff0c;在处理大规模数据抓取任务时展现出强大的能力。然而&#xff0c;随着业务规模的不断扩大和数据抓取需求的日益复杂&#xff0c;传统…...

《信号与系统》第 6 章 信号与系统的时域和频域特性

目录 6.0 引言 6.1 傅里叶变换的模和相位表示 6.2 线性时不变系统频率响应的模和相位表示 6.2.1 线性与非线性相位 6.2.2 群时延 6.2.3 对数模和相位图 6.3 理想频率选择性滤波器的时域特性 6.4 非理想滤波器的时域和频域特性讨论 6.5 一阶与二阶连续时间系统 6.5.1 …...

前端工具库lodash与lodash-es区别详解

lodash 和 lodash-es 是同一工具库的两个不同版本&#xff0c;核心功能完全一致&#xff0c;主要区别在于模块化格式和优化方式&#xff0c;适合不同的开发环境。以下是详细对比&#xff1a; 1. 模块化格式 lodash 使用 CommonJS 模块格式&#xff08;require/module.exports&a…...