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

RocketMQ快速入门:事务消息原理及实现(十)

目录

  • 0. 引言
  • 1. 原理
  • 2. 事务消息的实现
    • 2.1 java client实现(适用于spring框架)
    • 2.2 springboot实现
  • 3. 总结

0. 引言

rocketmq 的一大特性就是支持事务性消息,这在诸多场景中有所应用。在之前的文章中我们已经讲解过事务消息的使用,但事务消息是如何实现的呢? 今天我们就来探究其原理

1. 原理

rocketmq事务消息的基本流程分为三步:
(1)发送 half 消息到 broker。
(2)执行本地事务。
(3)根据本地事务执行结果:本地事务执行成功则将half消息标识为COMMIT,将消息进行提交,变成普通消息,可以被消费者消费;本地事务执行失败将half消息标识为ROLLBACK,并将消息从消息日志中删除,消费者则消费不到了。

从流程上可以看出rocketmq的事务消息,和数据库的事务有些类型,都有一个二阶段提交的概念。

rocketmq的half message(半消息)是一种特殊的消息,该种消息不会被消费者消费,但是可以被TransactionListener事务监听器获取到。

而我们就在事务监听器中书写自己的本地事务逻辑,本地事务执行成功后才将消息正常提交,这时提交后的消息才能被消费者消费到,否则就会回滚消息,消息就相当于从来没发送过。

这里需要注意的是,因为我们的本地事务逻辑已经在半消息接收的时候处理掉了,所以如果后续没有消费逻辑了,就不用再在消费者里书写逻辑,但如果还有后续的逻辑,就可以按照正常消费流程进行书写。
在这里插入图片描述

但是还有一个场景我们需要考虑,当本地事务执行完成后,在返回“执行成功(提交)”或“执行失败(回滚)”状态时,因为网络波动或者broker服务挂了,导致broker没有正常收到这个状态,从而无法及时把half message进行提交或回滚

这时就需要有个定时巡查机制,来检查这些没有正常收到提交状态的消息的实际状态到底是什么,这个巡查机制就是消息回查,也称为事务消息的补偿。在事务监听器TransactionListener中就是通过checkLocalTransaction方法来实现,executeLocalTransaction方法返回值一共有3种状态:

  • COMMIT_MESSAGE

提交状态,事务正常进行,一般是本地事务执行成功后进行设置。告知broker提交该事务消息,然后消费者可以消费该消息,当然此时消费者已经执行完本地事务了,再消费可以根据业务逻辑进行后续的逻辑处理,如果没有相关逻辑了忽略消息即可

  • ROLLBACK_MESSAGE

回滚状态,事务撤回,broker将删除当前half消息,一般是本地事务执行失败后进行设置

  • UNKNOW

未知状态,固定时间后Broker端会通过checkLocalTransaction方法进行消息回查,根据回查结果来判断该消息是提交还是回滚

示例代码:

@Overridepublic LocalTransactionState executeLocalTransaction(Message msg, Object arg) {// TODO 执行本地事务(书写你自己的本地事务逻辑)try{String body = new String(msg.getBody());int i = Integer.parseInt(body);// 模拟偶数执行成功,奇数执行失败if(i % 2 == 0){System.out.println("本地事务执行成功:"+body);// 执行成功return LocalTransactionState.COMMIT_MESSAGE;}else{System.out.println("本地事务执行失败:"+body);// 执行失败return LocalTransactionState.ROLLBACK_MESSAGE;}}catch (Exception e){e.printStackTrace();// 执行失败return LocalTransactionState.ROLLBACK_MESSAGE;}}

因此checkLocalTransaction方法中就要书写检查本地事务状态的方法,比如事务是对订单提交消息的消费,那么就去查询订单状态,如果已经是提交状态那么就返回COMMIT_MESSAGE,否则就返回ROLLBACK_MESSAGE

示例代码

       @Overridepublic LocalTransactionState checkLocalTransaction(MessageExt msg) {// TODO 去缓存或者数据库查询当前消息的实际状态// 模拟查询到状态为1Integer status = 1;// 不同实际状态对应的消息状态if (null != status) {switch (status) {case 1:return LocalTransactionState.COMMIT_MESSAGE;case 2:return LocalTransactionState.ROLLBACK_MESSAGE;default:return LocalTransactionState.COMMIT_MESSAGE;}}return LocalTransactionState.COMMIT_MESSAGE;}

2. 事务消息的实现

2.1 java client实现(适用于spring框架)

参考之前文章:https://wu55555.blog.csdn.net/article/details/138338692

2.2 springboot实现

参考之前文章:https://wu55555.blog.csdn.net/article/details/139741449

3. 总结

最后一句话总结一下,rocketmq的事务消息,是通过half消息(即二阶段提交)+回查机制来实现的。

相关文章:

RocketMQ快速入门:事务消息原理及实现(十)

目录 0. 引言1. 原理2. 事务消息的实现2.1 java client实现(适用于spring框架)2.2 springboot实现 3. 总结 0. 引言 rocketmq 的一大特性就是支持事务性消息,这在诸多场景中有所应用。在之前的文章中我们已经讲解过事务消息的使用&#xff0…...

Kotlin设计模式:深入理解桥接模式

Kotlin设计模式:深入理解桥接模式 在软件开发中,随着系统需求的不断增长和变化,类的职责可能会变得越来越复杂,导致代码难以维护和扩展。桥接模式(Bridge Pattern)是一种结构型设计模式,它通过…...

常用MQ消息中间件Kafka、ZeroMQ和RabbitMQ对比及RabbitMQ详解

1、概述 在现代的分布式系统和实时数据处理领域,消息中间件扮演着关键的角色,用于解决应用程序之间的通信和数据传递的挑战。在众多的消息中间件解决方案中,Kafka、ZeroMQ和RabbitMQ 是备受关注和广泛应用的代表性系统。它们各自具有独特的特…...

【UE5.3】笔记6-第一个简单小游戏

打砖块小游戏: 1、制造一面砖块组成的墙 在关卡中放置一个cube,放这地面上,将其转换成蓝图类,改名BP_Cube,更换砖块的贴图,按住alt键进行拷贝,堆出一面墙,复制出来的会很多,全选移动…...

LeetCode---402周赛

题目列表 3184. 构成整天的下标对数目 I 3185. 构成整天的下标对数目 II 3186. 施咒的最大总伤害 3187. 数组中的峰值 一、构成整天的下标对数目 I & II 可以直接二重for循环暴力遍历出所有的下标对,然后统计符合条件的下标对数目返回。代码如下 class So…...

循环冗余校验

循环冗余校验(Cyclic Redundancy Check,简称CRC)是一种广泛使用的错误检测编码技术,用于检测数据在传输或存储过程中是否发生错误。CRC通过在数据后面添加一个校验值(通常称为CRC码或CRC校验和)来实现错误检…...

resample sensor

resample sensor 的一个问题。 背景: 项目要求,发送多个数据到 sensor-hal 上去,发现无论怎样,在 sensor-hal 上都 只有一个数据。 resample sensor 是重新采样,这个怎么理解的,我的理解是: 假设 sensor 采…...

【Linux】多线程的相关知识点

一、线程安全 1.1 可重入 VS 线程安全 1.1.1 概念 线程安全:多个线程并发执行同一段代码时,不会出现不同的结果。常见对全局变量或者静态变量进行操作,并且没有锁的保护的情况下,会出现问题。重入:同一个函数被不同…...

Java反射详解

Java反射 一.什么是反射 我们使用的一些像框架,tomcat,或者一些其他的组件(jackson 对象–>json)。他们可以做到给他什么类名,就可以创建给定类的对象,并调用该对象的方法和属性。这是如何做到的? 当他们加载我们…...

Spring Boot与Apache Kafka集成的深度指南

Spring Boot与Apache Kafka集成的深度指南 大家好,我是免费搭建查券返利机器人省钱赚佣金就用微赚淘客系统3.0的小编,也是冬天不穿秋裤,天冷也要风度的程序猿! 在现代分布式系统中,消息队列的作用愈发重要&#xff0…...

甄选版“论软件系统架构评估”,软考高级论文,系统架构设计师论文

论文真题 对于软件系统,尤其是大规模的复杂软件系统来说,软件的系统架构对于确保最终系统的质量具有十分重要的意义,不恰当的系统架构将给项目开发带来高昂的代价和难以避免的灾难。对一个系统架构进行评估,是为了:分析现有架构存在的潜在风险,检验设计中提出的质量需求,…...

uniapp开发企业微信内部应用

最近一直忙着开发项目,终于1.0版本开发完成,抽时间自己总结下在项目开发中遇到的技术点。此次项目属于自研产品,公司扩展业务,需要在企业微信中开发内部应用。因为工作中使用的是钉钉,很少使用企业微信,对于…...

0122__linux之eventfd理解

linux之eventfd理解-CSDN博客 Linux fd 系列 — eventfd 是什么?-CSDN博客...

数学建模 —— 查找数据

目录 百度搜索技巧 完全匹配搜索:查询词的外边加上双引号“ ” 标题必含关键词:查询词前加上intitle: 搜索文档:空格再输入filetype:文件格式 去掉不想要的:查询词后面加空格后加减号与关键字 知网查文献 先看知网的硕博士…...

合并有序链表

合并有序链表 图解代码如下 图解 虽然很复杂,但能够很好的理解怎么使用链表,以及对链表的指针类理解 代码如下 Node* merge_list_two_pointer(List& list1, List& list2) {Node* new_head1 list1.head;Node* new_head2 list2.head;Node* s…...

【SpringBoot Web框架实战教程】05 Spring Boot 使用 JdbcTemplate 操作数据库

不积跬步,无以至千里;不积小流,无以成江海。大家好,我是闲鹤,微信:xxh_1459,十多年开发、架构经验,先后在华为、迅雷服役过,也在高校从事教学3年;目前已创业了…...

Spark基于DPU的Native引擎算子卸载方案

1.背景介绍 Apache Spark(以下简称Spark)是一个开源的分布式计算框架,由UC Berkeley AMP Lab开发,可用于批处理、交互式查询(Spark SQL)、实时流处理(Spark Streaming)、机器学习&a…...

Mini2440 start.s 修改支持串口输出,方便调试 (四)

经常会遇到点板子的时候,板子没有任何反应!怎么知道板子有没有在正常启动,在uboot阶段 start.s 中加入串口打印信息是很有必要的! 输出串口信息 ***UART:mini-2440-uBoot*** ***UART:mini-2440-uBoot*** ***UART:mini-2440-uBoo…...

【教程】几种不同的RBF神经网络

本站原创文章,转载请说明来自《老饼讲解-机器学习》www.bbbdata.com 目录 一、经典RBF神经网络1.1.经典径向基神经网络是什么1.2.经典径向基神经网络-代码与示例 二、广义回归神经网络GRNN2.1.广义回归神经网络是什么2.2.广义回归神经网络是什么-代码与示例 三、概率…...

【Liunx-后端开发软件安装】Liunx安装FDFS并整合nginx

【Liunx-后端开发软件安装】Liunx安装nacos 文章中涉及的相关fdfs相关软件安装包请点击下载: https://download.csdn.net/download/weixin_49051190/89471122 一、简介 FastDFS是一个开源的轻量级分布式文件系统,它对文件进行管理,功能包括…...

vue3 定时器-定义全局方法 vue+ts

1.创建ts文件 路径&#xff1a;src/utils/timer.ts 完整代码&#xff1a; import { onUnmounted } from vuetype TimerCallback (...args: any[]) > voidexport function useGlobalTimer() {const timers: Map<number, NodeJS.Timeout> new Map()// 创建定时器con…...

【Web 进阶篇】优雅的接口设计:统一响应、全局异常处理与参数校验

系列回顾&#xff1a; 在上一篇中&#xff0c;我们成功地为应用集成了数据库&#xff0c;并使用 Spring Data JPA 实现了基本的 CRUD API。我们的应用现在能“记忆”数据了&#xff01;但是&#xff0c;如果你仔细审视那些 API&#xff0c;会发现它们还很“粗糙”&#xff1a;有…...

IoT/HCIP实验-3/LiteOS操作系统内核实验(任务、内存、信号量、CMSIS..)

文章目录 概述HelloWorld 工程C/C配置编译器主配置Makefile脚本烧录器主配置运行结果程序调用栈 任务管理实验实验结果osal 系统适配层osal_task_create 其他实验实验源码内存管理实验互斥锁实验信号量实验 CMISIS接口实验还是得JlINKCMSIS 简介LiteOS->CMSIS任务间消息交互…...

什么是Ansible Jinja2

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

HDFS分布式存储 zookeeper

hadoop介绍 狭义上hadoop是指apache的一款开源软件 用java语言实现开源框架&#xff0c;允许使用简单的变成模型跨计算机对大型集群进行分布式处理&#xff08;1.海量的数据存储 2.海量数据的计算&#xff09;Hadoop核心组件 hdfs&#xff08;分布式文件存储系统&#xff09;&a…...

基于Java Swing的电子通讯录设计与实现:附系统托盘功能代码详解

JAVASQL电子通讯录带系统托盘 一、系统概述 本电子通讯录系统采用Java Swing开发桌面应用&#xff0c;结合SQLite数据库实现联系人管理功能&#xff0c;并集成系统托盘功能提升用户体验。系统支持联系人的增删改查、分组管理、搜索过滤等功能&#xff0c;同时可以最小化到系统…...

使用Spring AI和MCP协议构建图片搜索服务

目录 使用Spring AI和MCP协议构建图片搜索服务 引言 技术栈概览 项目架构设计 架构图 服务端开发 1. 创建Spring Boot项目 2. 实现图片搜索工具 3. 配置传输模式 Stdio模式&#xff08;本地调用&#xff09; SSE模式&#xff08;远程调用&#xff09; 4. 注册工具提…...

腾讯云V3签名

想要接入腾讯云的Api&#xff0c;必然先按其文档计算出所要求的签名。 之前也调用过腾讯云的接口&#xff0c;但总是卡在签名这一步&#xff0c;最后放弃选择SDK&#xff0c;这次终于自己代码实现。 可能腾讯云翻新了接口文档&#xff0c;现在阅读起来&#xff0c;清晰了很多&…...

【 java 虚拟机知识 第一篇 】

目录 1.内存模型 1.1.JVM内存模型的介绍 1.2.堆和栈的区别 1.3.栈的存储细节 1.4.堆的部分 1.5.程序计数器的作用 1.6.方法区的内容 1.7.字符串池 1.8.引用类型 1.9.内存泄漏与内存溢出 1.10.会出现内存溢出的结构 1.内存模型 1.1.JVM内存模型的介绍 内存模型主要分…...

十九、【用户管理与权限 - 篇一】后端基础:用户列表与角色模型的初步构建

【用户管理与权限 - 篇一】后端基础:用户列表与角色模型的初步构建 前言准备工作第一部分:回顾 Django 内置的 `User` 模型第二部分:设计并创建 `Role` 和 `UserProfile` 模型第三部分:创建 Serializers第四部分:创建 ViewSets第五部分:注册 API 路由第六部分:后端初步测…...