rocketmq源码分析4-事务消息实现原理

2022-09-23 14:20:20

为什么消息要具备事务能力

参见
还是比较清晰的。简单的说就是在你业务逻辑过程中，需要发送一条消息给订阅消息的人，但是期望是此逻辑过程完全成功完成之后才能使订阅者收到消息。
业务逻辑过程假设是这样的：
逻辑部分a-->发消息给MQ-->逻辑部分b
假设我们在发送消息给MQ之后执行逻辑部分b时产生了异常，那如果MQ不具备事务消息能力时，订阅者也收到了消息。这是我们不希望见到的。

分布式事务基础概念

rocketmq具备事务能力的demo

参见TransactionProducerDemo.java

向producer注册的TransactionCheckListener实现并没有用，因为返回LocalTransactionState.UNKNOW状态时，在3.2.6版本中，是不支持此状态下回调TransactionCheckListener的，具体参见以下两个issue。

事务消息 LocalTransactionState.UNKNOW 状态下不回查 #221
开源版本支持事务消息吗 #364
测试过程中发现返回UNKNOW状态是不能正确达到期望的，但是返回ROLLBACK_MESSAGE状态还是能达到期望的。

实现分析入口

这个实现的入口还是比较容易找的，只要搜寻ROLLBACK_MESSAGE这个变量的引用即可发现。顺着搜索查看，其实很容易发现，客户端在收到业务逻辑返回的事务状态时会发送一条结束事务的指令给broker。

// com.alibaba.rocketmq.client.impl.MQClientAPIImpl.endTransactionOneway(String, EndTransactionRequestHeader, String, long) 871行

RemotingCommand request =

                RemotingCommand.createRequestCommand(RequestCode.END_TRANSACTION, requestHeader);

按broker对外部指令的常规做法，一般会有一个Processor与之对应。是EndTransactionProcessor，看BrokerController374行其注册的地方，没错。

EndTransactionProcessor分析(broker侧)

如果LocalTransactionExecuter.executeLocalTransactionBranch返回LocalTransactionState.ROLLBACK_MESSAGE时，EndTransactionProcessor会清空message的body的置成null，queueOffset也不会更新，那么consumer就收不到消息了。

//--EndTransactionProcessor.processRequest  200行--

if (MessageSysFlag.TransactionRollbackType == requestHeader.getCommitOrRollback()) {

    msgInner.setBody(null);

}

如果LocalTransactionExecuter.executeLocalTransactionBranch返回LocalTransactionState.COMMIT_MESSAGE，那么EndTransactionProcessor则会照常put message。

事务消息分为两个阶段，prepare阶段与commit阶段。prepare阶段的消息会写入store，只是写完之后的queueOffset(逻辑位置)为0（commit阶段写完消息后的queueOffset就不是0了。）；

// -- com.alibaba.rocketmq.store.CommitLog.DefaultAppendMessageCallback.doAppend(long, ByteBuffer, int, Object) 1002行 --

final int tranType = MessageSysFlag.getTransactionValue(msgInner.getSysFlag());

switch (tranType) {

// Prepared and Rollback message is not consumed, will not enter the

// consumer queue

case MessageSysFlag.TransactionPreparedType:

case MessageSysFlag.TransactionRollbackType:

    queueOffset = 0L;

    break;

case MessageSysFlag.TransactionNotType:

case MessageSysFlag.TransactionCommitType:

default:

    break;

待分析问题列表：
1. prepare阶段已经将消息发了过去，commit的时候是否还会再发送一次消息？
2. rollback的时候是否会将prepare的消息删除？

码农公寓

为什么消息要具备事务能力

分布式事务基础概念

rocketmq具备事务能力的demo

实现分析入口

EndTransactionProcessor分析(broker侧)

相关文章