RocketMQ特性Broker存储事务消息实现
作者:奔跑的毛球 时间:2022-07-10 20:42:13
引言
在Broker中,事务消息的初始化是通过BrokerController.initialTransaction()
方法执行的。
private void initialTransaction() {
this.transactionalMessageService = ServiceProvider.loadClass(ServiceProvider.TRANSACTION_SERVICE_ID, TransactionalMessageService.class);
if (null == this.transactionalMessageService) {
this.transactionalMessageService = new TransactionalMessageServiceImpl(new TransactionalMessageBridge(this, this.getMessageStore()));
LOG.warn("Load default transaction message hook service: {}", TransactionalMessageServiceImpl.class.getSimpleName());
}
this.transactionalMessageCheckListener = ServiceProvider.loadClass(ServiceProvider.TRANSACTION_LISTENER_ID, AbstractTransactionalMessageCheckListener.class);
if (null == this.transactionalMessageCheckListener) {
this.transactionalMessageCheckListener = new DefaultTransactionalMessageCheckListener();
LOG.warn("Load default discard message hook service: {}", DefaultTransactionalMessageCheckListener.class.getSimpleName());
}
this.transactionalMessageCheckListener.setBrokerController(this);
this.transactionalMessageCheckService = new TransactionalMessageCheckService(this);
}
这里有三个核心的初始化变量
TransactionalMessageService
事务消息主要处理服务。默认实现类是TransactionalMessageServiceImpl
也可以自己定义事务消息处理实现类,通过ServiceProvider.loadClass()
方法进行加载。
TransactionalMessageService
类定义如下。内部属性已加注释标明。
public interface TransactionalMessageService {
//用于保存Half事务消息
PutMessageResult prepareMessage(MessageExtBrokerInner messageInner);
CompletableFuture<PutMessageResult> asyncPrepareMessage(MessageExtBrokerInner messageInner);
//删除事务消息
boolean deletePrepareMessage(MessageExt messageExt);
//提交事务消息
OperationResult commitMessage(EndTransactionRequestHeader requestHeader);
//回滚事务消息
OperationResult rollbackMessage(EndTransactionRequestHeader requestHeader);
void check(long transactionTimeout, int transactionCheckMax, AbstractTransactionalMessageCheckListener listener);
//打开事务消息
boolean open();
//关闭事务消息
void close();
}
transactionalMessageCheckListener
事务消息回查 *
transactionalMessageCheckService
事务消息回查服务,启动一个线程定时检查超时的Half消息是否需要回查。
处理事务消息
当初始化完成之后,Broker就可以处理事务消息了。
Broker存储事务消息的是org.apache.rocketmq.broker.processor.SendMessageProcessor
,这和普通消息其实是一样的。
但是有两点针对事务消息的特殊处理:
第一处:
在org.apache.rocketmq.broker.processor.SendMessageProcessor#sendMessage
中:
//获取扩展字段的值,若是该值为true则为事务消息
String traFlag = oriProps.get(MessageConst.PROPERTY_TRANSACTION_PREPARED);
boolean sendTransactionPrepareMessage = false;
if (Boolean.parseBoolean(traFlag)
&& !(msgInner.getReconsumeTimes() > 0 && msgInner.getDelayTimeLevel() > 0)) {
//判断当前Broker配置是否支持事务消息
if (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;
}
if (sendTransactionPrepareMessage) {
//保存Half信息
putMessageResult = this.brokerController.getTransactionalMessageService().prepareMessage(msgInner);
} else {
putMessageResult = this.brokerController.getMessageStore().putMessage(msgInner);
}
第二处:
存储事务消息前的预处理,对应方法是
org.apache.rocketmq.broker.transaction.queue.TransactionalMessageBridge#parseHalfMessageInner
private MessageExtBrokerInner parseHalfMessageInner(MessageExtBrokerInner msgInner) {
//将原消息的topic保存在扩展字段中
MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_REAL_TOPIC, msgInner.getTopic());
//将原消息的QueueId保存在扩展字段中
MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_REAL_QUEUE_ID,
String.valueOf(msgInner.getQueueId()));
//将原消息的SysFlag保存在扩展字段中
msgInner.setSysFlag(
MessageSysFlag.resetTransactionValue(msgInner.getSysFlag(), MessageSysFlag.TRANSACTION_NOT_TYPE));
//修改topic的值为RMQ_SYS_TRANS_HALF_TOPIC
msgInner.setTopic(TransactionalMessageUtil.buildHalfTopic());
//修改Queueid为0
msgInner.setQueueId(0);
msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties()));
return msgInner;
}
完成上述步骤之后,调用DefaultMessageStole.putMessage()
方法将其保存到CommitLog
中。
CommitLog存储成功之后,通过org.apache.rocketmq.store.CommitLog.DefaultAppendMessageCallback#doAppend()
方法对其进行处理。
final int tranType = MessageSysFlag.getTransactionValue(msgInner.getSysFlag());
switch (tranType) {
// Prepared and Rollback message is not consumed, will not enter the consume queue
case MessageSysFlag.TRANSACTION_PREPARED_TYPE:
case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE:
queueOffset = 0L;
break;
case MessageSysFlag.TRANSACTION_NOT_TYPE:
case MessageSysFlag.TRANSACTION_COMMIT_TYPE:
default:
break;
}
这里的逻辑是这样的,当读到的消息类型为事务消息时,设置当前消息的位点值为0,而不是设置真实的位点。这样该位点就不会建立ConsumeQueue索引,也不会被消费。
来源:https://juejin.cn/post/7132133887043436581
![](/images/zang.png)
![](/images/jiucuo.png)
猜你喜欢
JDK14之jpackage打包命令的使用
Eureka源码阅读之环境搭建及工程结构
![](https://img.aspxhome.com/file/2023/6/57946_0s.png)
java实现简单银行家算法
![](https://img.aspxhome.com/file/2023/7/63167_0s.jpg)
Spring boot集成Kafka消息中间件代码实例
![](https://img.aspxhome.com/file/2023/6/61786_0s.jpg)
springmvc前台向后台传值几种方式总结(从简单到复杂)
将JavaDoc注释生成API文档的操作
![](https://img.aspxhome.com/file/2023/1/61151_0s.png)
Spring Boot ActiveMQ连接池配置过程解析
Spring注解@Configuration与@Bean注册组件的使用详解
![](https://img.aspxhome.com/file/2023/2/64532_0s.png)
MyBatis学习教程(四)-如何快速解决字段名与实体类属性名不相同的冲突问题
C#设计模式之职责链模式示例详解
解析Java的Spring框架的BeanPostProcessor发布处理器
![](https://img.aspxhome.com/file/2023/0/65170_0s.png)
jvm细节探索之synchronized及实现问题分析
![](https://img.aspxhome.com/file/2023/6/57916_0s.png)
聊聊Java的switch为什么不支持long
基于Spring整合mybatis的mapper生成过程
![](https://img.aspxhome.com/file/2023/9/61129_0s.png)
Struts2 Result 参数详解
List集合对象中按照不同属性大小排序的实例
ThreadLocal使用案例_动力节点Java学院整理
Spring Boot 整合 Apache Dubbo的示例代码
![](https://img.aspxhome.com/file/2023/8/60488_0s.jpg)
Flutter 底部弹窗如何实现多项选择
![](https://img.aspxhome.com/file/2023/7/84707_0s.gif)
通过实例解析Spring Ioc项目实现过程
![](https://img.aspxhome.com/file/2023/5/59745_0s.png)