Go微服务: 分布式之发送带有事务消息的示例
分布式之发送带有事务消息
- 现在做一个RocketMQ的事务消息的 demo
1 )生产者
package mainimport ("context""fmt""time""github.com/apache/rocketmq-client-go/v2""github.com/apache/rocketmq-client-go/v2/primitive""github.com/apache/rocketmq-client-go/v2/producer"
)// 自定义结构体,为了实现 NewTransactionProducer 第一个参数的接口
type MyListener struct{}func (hl MyListener) ExecuteLocalTransaction(*primitive.Message) primitive.LocalTransactionState {return primitive.CommitMessageState
}func (hl MyListener) CheckLocalTransaction(*primitive.MessageExt) primitive.LocalTransactionState {return primitive.CommitMessageState
}func main() {// ------------ 1. 连接RocketMQ ------------------------mqAddr := "127.0.0.1:9876" // 模拟地址// NewTransactionProducer 这个方法第一个参数是一个 Listener// 是一个接口,需要一个接口体去实现它的方法p, err := rocketmq.NewTransactionProducer( // 开启事物消息生产者MyListener{},producer.WithNameServer([]string{mqAddr}),)if err != nil {panic(err) // 生产环境禁用panic}// ------------ 2. 启动RocketMQ ------------------------err = p.Start()if err != nil {panic(err)}// ------------ 3. 发送RocketMQ 消息 ------------------------res, err := p.SendMessageInTransaction(context.Background(),primitive.NewMessage("MyTransactionTopic", []byte("xxxxxxxxxxxyyyyyyyyyyyyyzzzzzzzzzzzz")),)fmt.Println(res.Status)if err != nil {panic(err)}fmt.Printf("发送成功")time.Sleep(time.Second * 3600)err = p.Shutdown()if err != nil {panic(err)}
}
- 可见,
primitive.LocalTransactionState
是返回值 - 进入这个包中,它有三个状态
CommitMessageState
提交状态RollbackMessageState
回滚状态UnknowState
未知状态
- 在后续,这三个状态都可以再试一试
2 ) 运行后,在UI界面查看消息
2.1 输出信息如下
INFO[0000] change the route for clients
INFO[0000] the topic route info changed changeTo="{\"OrderTopicConf\":\"\",\"queueDatas\":[{\"brokerName\":\"broker-bbs\",\"readQueueNums\":4,\"writeQueueNums\":4,\"perm\":6,\"topicSynFlag\":0}],\"brokerDatas\":[{\"cluster\":\"DefaultCluster\",\"brokerName\":\"broker-bbs\",\"brokerAddrs\":{\"0\":\"192.168.124.6:10911\"}}]}" changedFrom="<nil>" topic=MyTransactionTopic
0
发送成功
2.2 运行效果
2 )消费者
package mainimport ("context""fmt""time""github.com/apache/rocketmq-client-go/v2""github.com/apache/rocketmq-client-go/v2/consumer""github.com/apache/rocketmq-client-go/v2/primitive"
)func main() {mqAddr := "127.0.0.1:9876"topic := "MyTransactionTopic"groupName := "ddddddd"c, err := rocketmq.NewPushConsumer(consumer.WithGroupName(groupName),consumer.WithNsResolver(primitive.NewPassthroughResolver([]string{mqAddr})),)if err != nil {panic(err)}err = c.Subscribe(topic, consumer.MessageSelector{},func(ctx context.Context, msgList ...*primitive.MessageExt) (consumer.ConsumeResult, error) {for i := range msgList {fmt.Printf("订阅消息,消费%v \n", msgList[i])}return consumer.ConsumeSuccess, nil})if err != nil {fmt.Println("消费消息错误: %v", err.Error())}err = c.Start()if err != nil {fmt.Println("开启消费这错误: %v", err.Error())}time.Sleep(time.Hour)err = c.Shutdown()if err != nil {fmt.Println("shutdown消费者错误: %v", err.Error())}
}
-
在RocketMQ中,事务消息的处理机制涉及到生产者和消费者两端的协作,但与普通消息消费模式有所区别
-
事务消息的消费端并不直接参与到事务的两阶段提交过程中,它更像是一个“半事务消息”的确认者
-
具体流程如下:
- 生产者发送事务消息:生产者发送一条半事务消息到MQ服务器,并立即返回,此时消息处于“Prepare”状态
- MQ Server回调生产者确认:MQ服务器会回调生产者提供的事务监听器(在Go示例中是HappyListener),执行本地事务。生产者需在此阶段执行事务操作并决定是提交还是回滚该消息
- 生产者根据本地事务结果告知MQ Server:生产者根据本地事务执行结果,通过事务状态检查接口告诉MQ服务器是提交还是回滚这条半事务消息。
- 消息变为可消费状态:MQ服务器根据生产者的决定,将消息标记为Commit或Rollback,Commit后的消息才对普通消费者可见
-
因此,对于事务消息的消费者来说,其主要职责是消费那些已经被事务提交成功的消息,而不需要直接参与事务的提交或回滚过程
-
消费者代码看起来与普通消息的消费者相似,但消费的消息实际上是生产者已经提交成功的事务消息
-
不过,如果您的需求是希望消费者也以某种形式参与到事务的最终确认中,比如基于消息的消费结果来决定是否提交事务,这在RocketMQ的标准事务消息模型中并不直接支持
-
RocketMQ的事务模型主要关注于保证消息生产和本地事务的原子性,消费者更多的是作为事务结果的后续处理者角色
-
在提供的消费者示例中,尽管它看起来是一个普通的消费者,但实际上它处理的是生产者通过事务消息流程提交后的内容,这符合事务消息的消费逻辑
-
如果需要在消费端实现更复杂的逻辑来间接响应事务状态,可能需要结合业务系统进行额外的设计,比如通过监听数据库状态变化、消息队列的死信队列特性或其他补偿机制来处理未决事务
延迟性事务消息
- 您需要在创建消息时指定消息的延迟等级,而不是在生产者配置或消息发送后进行延迟
- RocketMQ支持多种延迟等级,每种等级对应不同的延迟时间
- 注意,事务消息和延迟消息的直接组合在RocketMQ中并不是直接支持的特性
- 因为事务消息的设计主要是围绕两阶段提交模型,确保消息发送与本地事务的一致性
- 而延迟消息侧重于消息的定时投递
- 然而,可以通过间接的方式结合这两个特性,即在事务消息的本地事务逻辑中包含对延迟操作的处理
- 下面的示例尝试模拟一种结合方式,但请注意,这仅是一种逻辑上的结合,实际应用中需要根据具体业务场景仔细设计和测试
- 方案思路:
生产者
:发送一个事务消息到特定的主题(例如DelayedTransactionTopic),该消息体中携带了需要进行延迟处理的信息。事务监听器
:在ExecuteLocalTransaction方法中,不直接执行长时间的延迟逻辑,而是执行快速操作(如记录消息待处理状态或存入DB),然后返回primitive.Prepared状态。检查事务状态
:在CheckLocalTransaction方法中,检查事务状态,如果需要,触发一个异步任务或消息队列中的消息,该消息携带延迟处理逻辑和真正的延迟时间。延迟处理服务
:这个服务从队列中取出消息并根据消息中的指示进行真正的延迟操作,例如通过内部队列或定时任务系统(如分布式定时任务框架)来实现延迟执行
- 相关生产者伪代码未完全实现,仅供参考
package mainimport ("context""fmt""time""github.com/apache/rocketmq-client-go/v2""github.com/apache/rocketmq-client-go/v2/primitive""github.com/apache/rocketmq-client-go/v2/producer" )type DelayedTxListener struct{}func (d DelayedTxListener) ExecuteLocalTransaction(msg *primitive.Message) primitive.LocalTransactionState {// 假设这里将消息标识为待处理,并记录相关信息到数据库fmt.Println("Preparing transaction, storing message meta...")return primitive.Prepared }func (d DelayedTxListener) CheckLocalTransaction(msgExt *primitive.MessageExt) primitive.LocalTransactionState {// 在这里检查消息是否准备好执行延迟操作// 实际操作可能包括从数据库查询该消息的状态// 假设我们已经确定需要进行延迟操作,这里直接模拟提交return primitive.CommitMessageState }func main() {mqAddr := "127.0.0.1:9876"p, err := rocketmq.NewTransactionProducer(DelayedTxListener{},producer.WithNameServer([]string{mqAddr}),)if err != nil {panic(err)}err = p.Start()if err != nil {panic(err)}msg := &primitive.Message{Topic: "DelayedTransactionTopic",Body: []byte("消息内容,可以包含延迟处理的详细信息"),}res, err := p.SendMessageInTransaction(context.Background(), msg)if err != nil {panic(err)}fmt.Printf("发送事务消息状态: %v\n", res.Status)time.Sleep(time.Second * 3600)err = p.Shutdown()if err != nil {panic(err)} }
- 注意事项
- 这个示例主要是概念性的,展示了如何在事务消息的上下文中计划后续的延迟处理步骤,而没有直接实现延迟消息的发送。
- 实际应用中,您可能需要实现一个额外的后台服务或消息队列来处理这些“计划”好的延迟任务,确保它们能在预定时间得到执行。
- 事务消息的两阶段提交机制仍然适用,只是延迟操作本身不在RocketMQ的直接事务控制范围内,而是作为一种业务逻辑上的后续处理。
- 务必根据您的具体业务需求和RocketMQ的版本特性,仔细设计和测试这样的解决方案。
相关文章:
Go微服务: 分布式之发送带有事务消息的示例
分布式之发送带有事务消息 现在做一个RocketMQ的事务消息的 demo 1 )生产者 package mainimport ("context""fmt""time""github.com/apache/rocketmq-client-go/v2""github.com/apache/rocketmq-client-go/v2/prim…...
【go】go初始化命令总结
包初始化 test项目目录下执行 go mod init test go mod tidy生成二进制可执行文件 go build -o test .\main.go...
vue音乐播放条
先看效果 再看代码 <template><div class"footer-player z-30 flex items-center p-2"><div v-if"isShow" class"h-12 w-60 overflow-hidden"><div :style"activeStyle" class"open-detail-control-wrap&…...
halcon实现浓淡补正,中间值补正-抽取暗
代码效果 抽取前 中值抽取暗 halcon函数代码 测试图片参数 NoiseCut:16 Gain:1 输入ImagePart NoiseCut Gain *获取直方图 get_domain (ImagePart, Domain) gray_histo_range(Domain,ImagePart,0,255,256, Histo, BinSize) area_center(Domain, NumPixels, Row, Column) …...
太速科技-FMC213V3-基于FMC兼容1.8V IO的Full Camera Link 输入子卡
FMC213V3-基于FMC兼容1.8V IO的Full Camera Link 输入子卡 一、板卡概述 该板卡为了考虑兼容1.8V电平IO,适配Virtex7,Kintex Ultrascale,Virtex ultrasacle FPGA而特制,如果要兼容原来的3.3V 也可以修改硬件参数。板卡支持1路…...
GPU短缺和模型效率的推动
1. 引言 随着全球GPU短缺和云计算成本的不断上升,开发更高效的AI模型成为了当前的焦点。技术如低秩适应(LoRA)和量化(Quantization)在优化性能的同时,减少了资源需求。这些技术不仅在当前的AI开发中至关重…...
linux在文件夹中查找文件内容
linux在文件夹中查找文件内容 在Linux中,可以通过以下多个途径,在文件夹中查找文件内容: 1、使用grep命令: grep -r "要查找的内容" /path/to/folder-r参数表示递归地在文件夹及其子文件夹中搜索。/path/to/folder是要搜索的文件夹路径。2、使用ack命令 ack …...
算法:11. 盛最多水的容器
11. 盛最多水的容器 给定一个长度为 n 的整数数组 height 。有 n 条垂线,第 i 条线的两个端点是 (i, 0) 和 (i, height[i]) 。 找出其中的两条线,使得它们与 x 轴共同构成的容器可以容纳最多的水。 返回容器可以储存的最大水量。 说明:你…...
Hazelcast 分布式缓存 在Seatunnel中的使用
1、背景 最近在调研seatunnel的时候,发现新版的seatunnel提供了一个web服务,可以用于图形化的创建数据同步任务,然后管理任务。这里面有个日志模块,可以查看任务的执行状态。其中有个取读数据条数和同步数据条数。很好奇这个数据…...
分数限制下,选好专业还是选好学校?
目录 分数限制下,选好专业还是选好学校? 方向一:专业解析 1. 专业选择的重要性 2. 不同专业的优势与挑战 3. 个人专业选择经验分享 4. 实际场景下的“专业VS学校”选择方案 方向二:名校效应分析 1. 名校声誉与品牌效应 2…...
软件改为开机自启动
1.按键 win R,输入“shell:startup”命令, 然后就可以打开启动目录了,如下: 2.然后,把要开机启动的程序的图标拖进去即可。 参考:开机启动项如何设置...
集群down机的应急和恢复测试(非重做备机)
1. 集群的两台服务器的状态 实例 正常情况主备 ip 端口 node1 主机 192.168.6.6 9088 node2 备机 192.168.6.7 9088 2. 测试的步骤 down掉node1观察node2的状态在node2未自动切换的时候手动将node2调整为单机状态,模拟紧急使用模拟不紧急时࿰…...
【数据库系统概论复习】关系数据库与关系代数笔记
文章目录 基本概念数据库基本概念关系数据结构完整性约束 关系代数关系代数练习课堂练习 语法树 基本概念 数据库基本概念 DB 数据库, 为了存用户的各种数据,我们要建很多关系(二维表),所以把相关的关系(二…...
赛氪网受邀参加上海闵行区翻译协会年会,共探科技翻译创新之路
在科技飞速发展的时代背景下,翻译行业正面临着前所未有的机遇与挑战。作为连接高校、企业与社会的桥梁,赛氪网在推动翻译创新、促进学术交流方面展现出了独特的魅力。2024年6月9日,在华东师范大学外语学院举办的第十三届上海市闵行区翻译协会…...
项目管理进阶之EVM(挣值管理)
前言 项目管理进阶系列,终于有时间更新啦!!!欢迎持续关注哦~ 上一节博主重点讲了一个环:PDCA,无论各行各业,上到航空航天、下到种地种菜,都离不开对质量的监督和改进。这个环既是一…...
PLSQL、Oracle以及客户端远程连接服务器笔记(仅供参考)
1.PLSQL参考链接: 全网最全最细的PLSQL下载、安装、配置、使用指南、问题解答,相关问题已汇总-CSDN博客文章浏览阅读2.9w次,点赞98次,收藏447次。双击之后,这里选择安装目录,你安装目录选的哪里࿰…...
Win快速删除node_modules
在Windows系统上删除 node_modules 文件夹通常是一个缓慢且耗时的过程。这主要是由于几个关键因素导致的: 主要原因 文件数量多且嵌套深: node_modules 文件夹通常包含成千上万的子文件夹和文件。由于其结构复杂,文件和文件夹往往嵌套得非常…...
【机器学习】基于顺序到顺序Transformer机器翻译
引言 1.1 序列到序列模型详解 序列到序列(Seq2Seq)模型是深度学习中处理序列数据转换问题的关键架构。在自然语言处理(NLP)任务中,如机器翻译、文本摘要和聊天机器人等,Seq2Seq模型能够高效地将输入序列转换为期望的输出序列。 模型架构: 编…...
TEA 加密的 Java 实现
import java.nio.ByteBuffer; import java.nio.ByteOrder;public class TeaUtils {private static final int DELTA 0x9E3779B9;private static final int ROUND 32;private static final String KEY "password";/*** 加密字符串,使用 TEA 加密算法*/p…...
鸿蒙开发电话服务:【@ohos.telephony.data (蜂窝数据)】
蜂窝数据 说明: 本模块首批接口从API version 7开始支持。后续版本的新增接口,采用上角标单独标记接口的起始版本。 导入模块 import data from ohos.telephony.data;data.getDefaultCellularDataSlotId getDefaultCellularDataSlotId(callback: Async…...
Maven认识与学习
1. Maven介绍 1.2 初识Maven 1.2.1 什么是Maven Maven是Apache旗下的一个开源项目,是一款用于管理和构建java项目的工具。 官网:Maven – Welcome to Apache Maven Apache 软件基金会,成立于1999年7月,是目前世界上最大的最受…...
“深入探讨Redis主从复制:原理、配置与优化“
目录 # 概念 1. 配置主从同步步骤 1.1 创建文件夹 1.2 复制配置文件 1.3 配置文件关闭 1.4 查看端口号,发现端口号存在 1.5 连接三个端口号 1.6 查看主机运行情况 1.7 让服务器变成(主机)或(从机) 1.8 实现效…...
HTML初体验
可参考jd.com官网,ctrlu查看当前页面源代码 找到你的项目,在项目中创建html类型的网页文件 标准的HTML正确书写格式 <!DOCTYPE html> <html lang"en"> <head><meta charset"UTF-8"><title>Title&…...
全局特征提取netvlad的理解
...
【设计模式-12】代理模式的代码实现及使用场景
&emsp;代理模式是一种应用很广发的结构性设计模式,它的设计初衷就是通过引入新的代理对象,在客户端和目标对象之间起到中介的作用,从而实现控制客户端对目标对象的访问,比如增强或者阉割某些能力。 1. 概述 代理模…...
网工内推 | 神州数码、弧聚科技网工,IE认证优先,最高18K
01 神州数码 🔷招聘岗位:高级网络工程师 🔷岗位职责: 1)提供7*24小时一线运维技术服务,如因应急故障处理应15分钟内到达现场。 2)提供设备的告警信息的分析处理及与故障问题定位服务。 3)完成数据中心网络和HPC超算…...
【Linux】模拟实现一个简单的日志系统
👦个人主页:Weraphael ✍🏻作者简介:目前正在学习c和算法 ✈️专栏:Linux 🐋 希望大家多多支持,咱一起进步!😁 如果文章有啥瑕疵,希望大佬指点一二 如果文章对…...
MongoDB 多层级查询
多层级查询 注意:要注意代码顺序 查询层级数据代码放前面,查询条件放后面 if (StringUtils.isBlank(params.getDocType())) {params.setDocType(DOC_TDCTYPE);}String docName mapper.findByDocInfo(params.getDocType());List<ExpertApprovalOpin…...
grpc代理服务的实现(一)
最近公司需要无感知基于服务代号来实现通信, 并监控和管理通信连接,目前公司使用的是如下的逻辑(当然逻辑简化了,但是思想不变) 目录 简单的原理图代理服务的实现创建 tls tcp 服务, 用于grpc client 和 grpc service 通信保存 与 代理服务建立的 grpc …...
FastAPI系列 4 -路由管理APIRouter
FastAPI系列 -路由管理APIRouter 文章目录 FastAPI系列 -路由管理APIRouter一、前言二、APIRouter使用示例1、功能拆分2、users、books模块开发3、FastAPI主体 三、运行结果 一、前言 未来的py开发者请上座,在使用python做为后端开发一个应用程序或 Web API&#x…...
wordpress建个人网站/关于seo如何优化
canbeing加大个人形象投资,购买品牌眼镜2月3日,经过近一个月的观察、分析、比较,canbeing终于决定花1198元在“博士眼镜”购买眼镜一副。此眼镜配置如下:宝姿金属半框pt2317-sgn(镜架,1198元)、…...
太原网站建设注意/北京百度seo公司
熟练应用正则表达式,而且匹配的目标是纯文本,那么相比于写分析器来说,正则可以更快速的完成工作。还有在捕获字符串的能力,正则也可以很好的完成工作,比如截取url的域名或者其他的内容等等 正则表达式,听起…...
游戏推荐网站怎么做/一键优化表格
云开发(微信-小程序)笔记(十一)---- 分页,不简单啊 搜索的分类有单字段,多字段(或),多字段(并)搜索等 1.导入数据集 首先向数据库导入免费数据…...
织梦调用网站备案号/明星百度指数在线查询
1.常用且容易忘记的快捷键 iter 生成增强forxun 1.1.循环相关的快捷键 itar 生成array for代码块 itco 生成Collection迭代 iten 生成enumeration遍历 iter 生成增强forxun itit 生成iterator 迭代 itli 生成List的遍历 ittok 生成String token遍历 itve 生成Vector数组迭代 …...
北京网站建设公司册/电商运营入门基础知识
路径:在顶部菜单栏小眼睛按钮 全局变量跟环境变量作用域不同。全局变量不受当前环境的影响,只要是在同一个项目下,该变量都会生效 环境变量和全局变量的调用方法都是一样的 {{变量名}} 定义 目前通过脚本定义全局变量(由预/后执…...
可以做游戏可以视频约会的网站/福建百度seo排名点击软件
Free Ur Mind-推荐使用FreeMind工具 什么是MindMap? MindMap(被译成思维导图或心智图)是一种思维工具,由英国的记忆之父托尼-博赞发明。MindMap是一种新的思维模式,它将左脑的逻辑、顺序、条例、文字、数字,以及右脑的…...