RocketMQ事务消息 超时重发还是原来的消息吗？

以下面的一个demo例子来分析一下，探索RocketMQ事务消息原理。

    public static final String PRODUCER_GROUP = "tran-test";public static final String DEFAULT_NAMESRVADDR = "127.0.0.1:9876";public static final String TOPIC = "Test";public static void main(String[] args) throws Exception {TransactionListener transactionListener = new TransactionListener() {@Overridepublic LocalTransactionState executeLocalTransaction(Message msg, Object arg) {System.out.println(String.format("executeLocalTransaction: %s", msg.getTransactionId()));return LocalTransactionState.UNKNOW;}@Overridepublic LocalTransactionState checkLocalTransaction(MessageExt msg) {System.out.println(String.format("checkLocalTransaction: tranId=%s, commitLogOffset=%s, queueOffset=%s, msgId=%s",msg.getTransactionId(), msg.getCommitLogOffset(),msg.getQueueOffset(), msg.getMsgId()));return LocalTransactionState.UNKNOW;}};TransactionMQProducer producer = new TransactionMQProducer(PRODUCER_GROUP);producer.setTransactionListener(transactionListener);producer.setNamesrvAddr(DEFAULT_NAMESRVADDR);producer.start();Message msg = new Message(TOPIC, "test".getBytes());SendResult sendResult = producer.sendMessageInTransaction(msg, null);System.out.println(String.format("sendResult: tranId=%s, offsetMsgId=%s, queueOffset=%s msgId=%s",sendResult.getTransactionId(), sendResult.getOffsetMsgId(),sendResult.getQueueOffset(), sendResult.getMsgId()));CountDownLatch countDownLatch = new CountDownLatch(1);countDownLatch.await();}

executeLocalTransaction: C0DE00428BEC18B4AAC27F377B6E0000
sendResult: tranId=C0DE00428BEC18B4AAC27F377B6E0000, offsetMsgId=null, queueOffset=82 msgId=C0DE00428BEC18B4AAC27F377B6E0000
checkLocalTransaction: tranId=C0DE00428BEC18B4AAC27F377B6E0000, commitLogOffset=1315411, queueOffset=83, msgId=C0DE004200002A9F0000000000141253
checkLocalTransaction: tranId=C0DE00428BEC18B4AAC27F377B6E0000, commitLogOffset=1315805, queueOffset=84, msgId=C0DE004200002A9F00000000001413DD
checkLocalTransaction: tranId=C0DE00428BEC18B4AAC27F377B6E0000, commitLogOffset=1316199, queueOffset=85, msgId=C0DE004200002A9F0000000000141567
checkLocalTransaction: tranId=C0DE00428BEC18B4AAC27F377B6E0000, commitLogOffset=1316593, queueOffset=86, msgId=C0DE004200002A9F00000000001416F1
checkLocalTransaction: tranId=C0DE00428BEC18B4AAC27F377B6E0000, commitLogOffset=1316987, queueOffset=87, msgId=C0DE004200002A9F000000000014187B
checkLocalTransaction: tranId=C0DE00428BEC18B4AAC27F377B6E0000, commitLogOffset=1317381, queueOffset=88, msgId=C0DE004200002A9F0000000000141A05
checkLocalTransaction: tranId=C0DE00428BEC18B4AAC27F377B6E0000, commitLogOffset=1317775, queueOffset=89, msgId=C0DE004200002A9F0000000000141B8F
checkLocalTransaction: tranId=C0DE00428BEC18B4AAC27F377B6E0000, commitLogOffset=1318169, queueOffset=90, msgId=C0DE004200002A9F0000000000141D19
checkLocalTransaction: tranId=C0DE00428BEC18B4AAC27F377B6E0000, commitLogOffset=1318563, queueOffset=91, msgId=C0DE004200002A9F0000000000141EA3
checkLocalTransaction: tranId=C0DE00428BEC18B4AAC27F377B6E0000, commitLogOffset=1318957, queueOffset=92, msgId=C0DE004200002A9F000000000014202D
checkLocalTransaction: tranId=C0DE00428BEC18B4AAC27F377B6E0000, commitLogOffset=1319352, queueOffset=93, msgId=C0DE004200002A9F00000000001421B8
checkLocalTransaction: tranId=C0DE00428BEC18B4AAC27F377B6E0000, commitLogOffset=1319747, queueOffset=94, msgId=C0DE004200002A9F0000000000142343
checkLocalTransaction: tranId=C0DE00428BEC18B4AAC27F377B6E0000, commitLogOffset=1320142, queueOffset=95, msgId=C0DE004200002A9F00000000001424CE
checkLocalTransaction: tranId=C0DE00428BEC18B4AAC27F377B6E0000, commitLogOffset=1320537, queueOffset=96, msgId=C0DE004200002A9F0000000000142659
checkLocalTransaction: tranId=C0DE00428BEC18B4AAC27F377B6E0000, commitLogOffset=1320932, queueOffset=97, msgId=C0DE004200002A9F00000000001427E4

通过上述例子的输出结果可以发现，checkLocalTransaction中queueOffset、msgId都发生的变化。那么在broker中到底发生了什么呢。

事务消息原理

当客户端发送一个事务消息时，MessageConst.PROPERTY_TRANSACTION_PREPARED=“true” 标记这个消息是一个事务消息。

        SendResult sendResult = null;MessageAccessor.putProperty(msg, MessageConst.PROPERTY_TRANSACTION_PREPARED, "true");MessageAccessor.putProperty(msg, MessageConst.PROPERTY_PRODUCER_GROUP, this.defaultMQProducer.getProducerGroup());try {sendResult = this.send(msg);} catch (Exception e) {throw new MQClientException("send message Exception", e);}

broker在收到消息时会取出traFlag，如果traFlag=true消息将交给TransactionalMessageService处理

        String traFlag = oriProps.get(MessageConst.PROPERTY_TRANSACTION_PREPARED);boolean sendTransactionPrepareMessage = false;if (Boolean.parseBoolean(traFlag)&& !(msgInner.getReconsumeTimes() > 0 && msgInner.getDelayTimeLevel() > 0)) { //For client under version 4.6.1if (this.brokerController.getBrokerConfig().isRejectTransactionMessage()) {response.setCode(ResponseCode.NO_PERMISSION);response.setRemark("the broker[" + this.brokerController.getBrokerConfig().getBrokerIP1()+ "] sending transaction message is forbidden");return response;}sendTransactionPrepareMessage = true;}long beginTimeMillis = this.brokerController.getMessageStore().now();if (brokerController.getBrokerConfig().isAsyncSendEnable()) {CompletableFuture<PutMessageResult> asyncPutMessageFuture;if (sendTransactionPrepareMessage) {//处理事务消息asyncPutMessageFuture = this.brokerController.getTransactionalMessageService().asyncPrepareMessage(msgInner);} else {asyncPutMessageFuture = this.brokerController.getMessageStore().asyncPutMessage(msgInner);}

TransactionalMessageService在保存消息时会将原来的topic使用RMQ_SYS_TRANS_HALF_TOPIC来替换，原topic信息存放在properties中。这样在是先把消息保存下来，而不让Consumer立刻就能收到。
当收到TransactionMQProducer发来的COMMIT_MESSAGE时，再将消息从RMQ_SYS_TRANS_HALF_TOPIC取出替换成原来的topic写入。同时再向RMQ_SYS_TRANS_OP_HALF_TOPIC的topic中也写一份。
broker通过对比RMQ_SYS_TRANS_OP_HALF_TOPIC和RMQ_SYS_TRANS_HALF_TOPIC中是否同时存在来判断事务消息是否结束了。
当收到的不是COMMIT_MESSAGE而是UNKNOW时，TransactionalMessageCheckService会定时回调TransactionMQProducer#checkLocalTransaction查询本地事务状态，默认最多检查15次。

在这里插入图片描述

TransactionalMessageCheckService

TransactionalMessageCheckService是一个运行在broker中的一个线程，线程默认每1分钟执行一次来检测系统中超时的half事务消息并发起重试。

    @Overridepublic void check(long transactionTimeout, int transactionCheckMax,AbstractTransactionalMessageCheckListener listener) {try {String topic = TopicValidator.RMQ_SYS_TRANS_HALF_TOPIC;Set<MessageQueue> msgQueues = transactionalMessageBridge.fetchMessageQueues(topic);if (msgQueues == null || msgQueues.size() == 0) {log.warn("The queue of topic is empty :" + topic);return;}log.debug("Check topic={}, queues={}", topic, msgQueues);for (MessageQueue messageQueue : msgQueues) {long startTime = System.currentTimeMillis();//每个half queue都有一个对应的op queueMessageQueue opQueue = getOpQueue(messageQueue);//获取当前未完成的half queue的offsetlong halfOffset = transactionalMessageBridge.fetchConsumeOffset(messageQueue);//获取当前已完成的op queue的offsetlong opOffset = transactionalMessageBridge.fetchConsumeOffset(opQueue);log.info("Before check, the queue={} msgOffset={} opOffset={}", messageQueue, halfOffset, opOffset);if (halfOffset < 0 || opOffset < 0) {log.error("MessageQueue: {} illegal offset read: {}, op offset: {},skip this queue", messageQueue,halfOffset, opOffset);continue;}......// single threadint getMessageNullCount = 1;long newOffset = halfOffset;long i = halfOffset;long nextOpOffset = pullResult.getNextBeginOffset();int putInQueueCount = 0;int escapeFailCnt = 0;while (true) {if (System.currentTimeMillis() - startTime > MAX_PROCESS_TIME_LIMIT) {log.info("Queue={} process time reach max={}", messageQueue, MAX_PROCESS_TIME_LIMIT);break;}if (removeMap.containsKey(i)) {......} else {//从RMQ_SYS_TRANS_HALF_TOPIC取出half消息GetResult getResult = getHalfMsg(messageQueue, i);MessageExt msgExt = getResult.getMsg();if (msgExt == null) {if (getMessageNullCount++ > MAX_RETRY_COUNT_WHEN_HALF_NULL) {break;}............//是否需要丢弃消息if (needDiscard(msgExt, transactionCheckMax) || needSkip(msgExt)) {listener.resolveDiscardMsg(msgExt);newOffset = i + 1;i++;continue;}......//判断上次check是否超时boolean isNeedCheck = opMsg == null && valueOfCurrentMinusBorn > checkImmunityTime|| opMsg != null && opMsg.get(opMsg.size() - 1).getBornTimestamp() - startTime > transactionTimeout|| valueOfCurrentMinusBorn <= -1;if (isNeedCheck) {//超时if (!putBackHalfMsgQueue(msgExt, i)) {continue;}putInQueueCount++;log.info("Check transaction. real_topic={},uniqKey={},offset={},commitLogOffset={}",msgExt.getUserProperty(MessageConst.PROPERTY_REAL_TOPIC),msgExt.getUserProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX),msgExt.getQueueOffset(), msgExt.getCommitLogOffset());//重新给TransactionListener发起check请求listener.resolveHalfMsg(msgExt);..................if (newOffset != halfOffset) {transactionalMessageBridge.updateConsumeOffset(messageQueue, newOffset);}long newOpOffset = calculateOpOffset(doneOpOffset, opOffset);if (newOpOffset != opOffset) {transactionalMessageBridge.updateConsumeOffset(opQueue, newOpOffset);}

上述代码中有三个比较重要的细节，needDiscard、putBackHalfMsgQueue和listener.resolveHalfMsg。
needDiscard：从half queue取出来后判断消息的TRANSACTION_CHECK_TIMES属性是否大于15次。
小于15次，则TRANSACTION_CHECK_TIMES属性值+1。
大于15次，则从RMQ_SYS_TRANS_HALF_TOPIC中丢弃，通过listener.resolveDiscardMsg保存在TRANS_CHECK_MAX_TIME_TOPIC中交由人工处理。
putBackHalfMsgQueue：将消息重新插入一份到RMQ_SYS_TRANS_HALF_TOPIC，因为CommitLog的applyOnly特性不能修改原消息。所以需要重新apply消息导致queueOffset、commitLogOffset、msgId都发生了变化。

    private boolean putBackHalfMsgQueue(MessageExt msgExt, long offset) {PutMessageResult putMessageResult = putBackToHalfQueueReturnResult(msgExt);if (putMessageResult != null&& putMessageResult.getPutMessageStatus() == PutMessageStatus.PUT_OK) {msgExt.setQueueOffset(putMessageResult.getAppendMessageResult().getLogicsOffset());msgExt.setCommitLogOffset(putMessageResult.getAppendMessageResult().getWroteOffset());msgExt.setMsgId(putMessageResult.getAppendMessageResult().getMsgId());log.debug("Send check message, the offset={} restored in queueOffset={} "+ "commitLogOffset={} "+ "newMsgId={} realMsgId={} topic={}",offset, msgExt.getQueueOffset(), msgExt.getCommitLogOffset(), msgExt.getMsgId(),msgExt.getUserProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX),msgExt.getTopic());return true;

listener.resolveHalfMsg：通过回调resolveHalfMsg方法向TransactionMQProducer重发check。

    public void resolveHalfMsg(final MessageExt msgExt) {if (executorService != null) {executorService.execute(new Runnable() {@Overridepublic void run() {try {sendCheckMessage(msgExt);} catch (Exception e) {LOGGER.error("Send check message error!", e);}}});} else {LOGGER.error("TransactionalMessageCheckListener not init");}}public void sendCheckMessage(MessageExt msgExt) throws Exception {CheckTransactionStateRequestHeader checkTransactionStateRequestHeader = new CheckTransactionStateRequestHeader();checkTransactionStateRequestHeader.setCommitLogOffset(msgExt.getCommitLogOffset());checkTransactionStateRequestHeader.setOffsetMsgId(msgExt.getMsgId());checkTransactionStateRequestHeader.setMsgId(msgExt.getUserProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX));checkTransactionStateRequestHeader.setTransactionId(checkTransactionStateRequestHeader.getMsgId());checkTransactionStateRequestHeader.setTranStateTableOffset(msgExt.getQueueOffset());checkTransactionStateRequestHeader.setBname(brokerController.getBrokerConfig().getBrokerName());msgExt.setTopic(msgExt.getUserProperty(MessageConst.PROPERTY_REAL_TOPIC));msgExt.setQueueId(Integer.parseInt(msgExt.getUserProperty(MessageConst.PROPERTY_REAL_QUEUE_ID)));msgExt.setStoreSize(0);String groupId = msgExt.getProperty(MessageConst.PROPERTY_PRODUCER_GROUP);Channel channel = brokerController.getProducerManager().getAvailableChannel(groupId);if (channel != null) {//取出与broker相连的netty channel发送check消息brokerController.getBroker2Client().checkProducerTransactionState(groupId, channel, checkTransactionStateRequestHeader, msgExt);} else {LOGGER.warn("Check transaction failed, channel is null. groupId={}", groupId);}}

half消息示意图

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/121125.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

【CSS】CSS 属性计算过程

【CSS】CSS 属性计算过程

1. 概述 我们所书写的任何一个 HTML 元素&#xff0c;实际上都有完整的一整套 CSS 样式。如果没有修改某样式&#xff0c;大概率可能使用默认值。 例如&#xff1a; <!DOCTYPE html> <html lang"en"> <head><meta charset"UTF-8"&…
阅读更多...
unity 基于UGUI的无限动态滚动列表

unity 基于UGUI的无限动态滚动列表

基于UGUI的动态滚动列表&#xff0c;主要支持以下功能&#xff1a; 继承自UGUI的SrollRect&#xff0c;支持ScrollRect的所有功能&#xff1b; 使用对象池来管理列表元素&#xff0c;以实现列表元素的复用&#xff1b; 支持一行多个元素或一列多个元素&#xff1b; 可使用不…
阅读更多...
【MyBatis Plus】深入探索 MyBatis Plus 的条件构造器，自定义 SQL语句，Service 接口的实现

【MyBatis Plus】深入探索 MyBatis Plus 的条件构造器，自定义 SQL语句，Service 接口的实现

文章目录 前言一、条件构造器1.1 什么是条件构造器1.2 QueryWrapper1.3 UpdateWrapper1.4 LambdaWrapper 二、自定义 SQL 语句2.1 自定义 SQL 的基本用法2.2 自定义 SQL 实现多表查询 三、Service 接口3.1 对 Service 接口的认识3.2 实现 Service 接口3.3 实现增删改查功能3.4 …
阅读更多...
el-form那些事

el-form那些事

vue3element-plus el-form那些事 输入框后拼接文字 输入框后拼接文字 <el-form-item :label"t(location.locationLength)" prop"locationLength"><el-input v-model"form.locationLength" :placeholder"t(location.inputLocation…
阅读更多...
Windoes定时任务、设置定时重启系统

Windoes定时任务、设置定时重启系统

步骤一&#xff1a; 打开计算机管理 通过&#xff1a;control(控制面板&#xff09;或者compmgmt.msc(计算机管理&#xff09;打开程序 步骤二&#xff1a;打开——>系统工具 步骤三&#xff1a; 选择——>任务计划程序 步骤四&#xff1a; 可选择创建新文件命名&…
阅读更多...
正点原子嵌入式linux驱动开发——Linux PWM驱动

正点原子嵌入式linux驱动开发——Linux PWM驱动

PWM是很常用到功能&#xff0c;可以通过PWM来控制电机速度&#xff0c;也可以使用PWM来控制LCD的背光亮度。本章就来学习一下如何在Linux下进行PWM驱动开发。 PWM驱动解析 不在介绍PWM是什么了&#xff0c;直接进入使用。 给LCD的背光引脚输入一个PWM信号&#xff0c;这样就…
阅读更多...
记一次企业微信的（CorpID）和密钥（Secret）泄漏的利用案例

记一次企业微信的（CorpID）和密钥（Secret）泄漏的利用案例

文章目录 一、介绍二、利用过程1、获取AccessToken2、获取企业微信接口IP段3、获取企业微信回调IP段4、通过部门ID，查看返回的ID5、通过部门ID，查看用户列表6、通过便利ID，发现用户信息泄露，可以进行提交报告7、通过添加接口，添加企业账号8、登陆企业账号进行测试三、参考…
阅读更多...
【随机过程】布朗运动

【随机过程】布朗运动

这里写目录标题 Brownian motion Brownian motion The brownian motion 1D and brownian motion 2D functions, written with the cumsum command and without for loops, are used to generate a one-dimensional and two-dimensional Brownian motion, respectively. 使用cu…
阅读更多...
主动调度是如何发生的

主动调度是如何发生的

计算机主要处理计算、网络、存储三个方面。计算主要是 CPU 和内存的合作&#xff1b;网络和存储则多是和外部设备的合作&#xff1b;在操作外部设备的时候&#xff0c;往往需要让出 CPU&#xff0c;就像上面两段代码一样&#xff0c;选择调用 schedule() 函数。 上下文切换主要…
阅读更多...
Kafka - 3.x 图解Broker总体工作流程

Kafka - 3.x 图解Broker总体工作流程

文章目录 Zk中存储的kafka的信息Kafka Broker总体工作流程1. broker启动后向zk中注册2. Controller谁先启动注册&#xff0c;谁说了算3. 由选举出来的Controller监听brokers节点的变化4. Controller决定leader选举5. Controller将节点信息上传到Zk中6. 其他Controller从zk中同步…
阅读更多...
Fourier分析导论——第1章——Fourier分析的起源(E.M. Stein R. Shakarchi)

Fourier分析导论——第1章——Fourier分析的起源(E.M. Stein R. Shakarchi)

第 1 章 Fourier分析的起源 (The Genesis of Fourier Analysis) Regarding the researches of dAlembert and Euler could one not add that if they knew this expansion, they made but a very imperfect use of it. They were both persuaded that an arbitrary and d…
阅读更多...
jenkins配置gitlab凭据

jenkins配置gitlab凭据

下载Credentials Binding插件&#xff08;默认是已经安装了&#xff09; 在凭据配置里添加凭据类型 点击保存 Username with password&#xff1a; 用户名和密码 SSH Username with private 在凭据管理里面添加gitlab账号和密码 点击全局 点击添加凭据&#xff08;版本不同…
阅读更多...
Go RESTful API 接口开发

Go RESTful API 接口开发

文章目录 什么是 RESTful APIGo 流行 Web 框架-GinGo HelloWorldGin 路由和控制器Gin 处理请求参数生成 HTTP 请求响应Gin 的学习内容实战用 Gin 框架开发 RESTful APIOAuth 2.0接口了解用 Go 开发 OAuth2.0 接口示例 编程有一个准则——Don‘t Repeat Yourself&#xff08;不要…
阅读更多...
如何在Windows和Linux系统上监听文件夹的变动？

如何在Windows和Linux系统上监听文件夹的变动？

文章目录 如何在Windows和Linux系统上监听文件夹的变动&#xff1f;读写文件文件系统的操作缓冲和流文件改变事件 如何在Windows和Linux系统上监听文件夹的变动&#xff1f; libuv库实现了监听整个文件夹的修改。本文详细介绍libuv库文件读写和监听的的实现方法。libuv库开发了…
阅读更多...
Unity的碰撞检测(六)

Unity的碰撞检测(六)

温馨提示&#xff1a;本文基于前一篇“Unity的碰撞检测(五)”继续探讨两个游戏对象具备刚体的BodyType均为Dynamic&#xff0c;但是Collision Detection属性不同的碰撞检测&#xff0c;阅读本文则默认已阅读前文。 &#xff08;一&#xff09;测试说明 在基于两个游戏对象都具…
阅读更多...
TSINGSEE青犀省级高速公路视频上云联网方案：全面实现联网化、共享化、智能化

TSINGSEE青犀省级高速公路视频上云联网方案：全面实现联网化、共享化、智能化

一、需求背景 随着高速铁路的建设及铁路管理的精细化&#xff0c;原有的模拟安防视频监控系统已经不能满足视频监控需求&#xff0c;越来越多站点在建设时已开始规划高清安防视频监控系统。高速公路视频监控资源非常丰富&#xff0c;需要对其进行综合管理与利用。通过构建监控…
阅读更多...
荣电集团与钕希科技签署全面战略合作

荣电集团与钕希科技签署全面战略合作

10月26日&#xff0c;荣电集团&#xff08;以下简称荣电&#xff09;与钕希科技南京有限公司&#xff08;以下简称钕希科技&#xff09;今天在合肥市签署全面战略合作协议&#xff0c;联合进军混合现实&#xff08;Mixed Reality&#xff0c;以下简称MR&#xff09;空间计算高科…
阅读更多...
Java练习题2021-4

Java练习题2021-4

"某游戏公司设计了一个奖励活动&#xff0c;给N个用户(1≤N≤10^7)连续编号为1到N&#xff0c;依据用户的编号S发放奖励。 发放奖励规则为&#xff1a; 公司随机设定三个非零正整数x&#xff0c;y&#xff0c;z。 如果S同时是x、y的倍数&#xff0c;奖励2张卡片&#xff1…
阅读更多...
如何绘制【逻辑回归】中threshold参数的学习曲线

如何绘制【逻辑回归】中threshold参数的学习曲线

threshold参数的意义是通过筛选掉低于threshold的参数&#xff0c;来对逻辑回归的特征进行降维。 首先导入相应的模块&#xff1a; from sklearn.linear_model import LogisticRegression as LR from sklearn.datasets import load_breast_cancer from sklearn.model_selecti…
阅读更多...
内核进程的调度与进程切换

内核进程的调度与进程切换

进程被创建到了链表中&#xff0c;如何再进行进一步的调用和调用&#xff1f; 进程调度 void schedule(void)&#xff1b; 进程调度 switch_to(next); 进程切换函数 void schedule(void) {int i,next,c;struct task_struct ** p;/* check alarm, wake up any i…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023