事务消息实现

深入理解RocketMQ事务消息的实现，掌握事务消息的完整流程和状态回查机制，理解事务消息存储、事务消息消费等核心功能。

📋 事务消息概览

事务消息是RocketMQ提供的分布式事务解决方案，保证消息和本地事务的一致性。

🎯 一、事务消息流程

1.1 完整流程

事务消息流程图：

1.2 TransactionMQProducer

TransactionMQProducer.sendMessageInTransaction：

@Override
public TransactionSendResult sendMessageInTransaction(final Message msg,
        final LocalTransactionExecuter localTransactionExecuter, 
        final Object arg) throws MQClientException {
    TransactionListener transactionListener = getCheckListener();
    if (null == localTransactionExecuter && null == transactionListener) {
        throw new MQClientException("tranExecutor is null", null);
    }
    
    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);
    }
    
    LocalTransactionState localTransactionState = LocalTransactionState.UNKNOW;
    Throwable localException = null;
    switch (sendResult.getSendStatus()) {
        case SEND_OK: {
            try {
                if (localTransactionExecuter != null) {
                    localTransactionState = localTransactionExecuter.executeLocalTransactionBranch(
                            msg, arg);
                } else if (transactionListener != null) {
                    localTransactionState = transactionListener.executeLocalTransaction(msg, arg);
                }
                if (localTransactionState == null) {
                    localTransactionState = LocalTransactionState.UNKNOW;
                }
                
                if (localTransactionState != LocalTransactionState.COMMIT_MESSAGE) {
                    log.info("executeLocalTransactionBranch return {}", localTransactionState);
                    log.info(msg.toString());
                }
            } catch (Throwable e) {
                log.info("executeLocalTransactionBranch exception", e);
                localException = e;
            }
        }
        break;
        case FLUSH_DISK_TIMEOUT:
        case FLUSH_SLAVE_TIMEOUT:
        case SLAVE_NOT_AVAILABLE:
            localTransactionState = LocalTransactionState.ROLLBACK_MESSAGE;
            break;
        default:
            break;
    }
    
    try {
        this.endTransaction(sendResult, localTransactionState, localException);
    } catch (Exception e) {
        log.warn("local transaction execute " + localTransactionState + 
                ", but end broker transaction failed", e);
    }
    
    TransactionSendResult transactionSendResult = new TransactionSendResult();
    transactionSendResult.setSendStatus(sendResult.getSendStatus());
    transactionSendResult.setMessageQueue(sendResult.getMessageQueue());
    transactionSendResult.setMsgId(sendResult.getMsgId());
    transactionSendResult.setQueueOffset(sendResult.getQueueOffset());
    transactionSendResult.setTransactionId(sendResult.getTransactionId());
    transactionSendResult.setLocalTransactionState(localTransactionState);
    return transactionSendResult;
}

🎯 二、状态回查

2.1 回查机制

TransactionCheckListener：

public interface TransactionCheckListener {
    LocalTransactionState checkLocalTransactionState(final MessageExt msg);
}

2.2 回查实现

TransactionMessageService.check：

public 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) {
            return;
        }
        
        for (MessageQueue messageQueue : msgQueues) {
            long startTime = System.currentTimeMillis();
            MessageQueue opQueue = getOpQueue(messageQueue);
            long halfOffset = transactionalMessageBridge.fetchConsumeOffset(
                    messageQueue);
            long opOffset = transactionalMessageBridge.fetchConsumeOffset(opQueue);
            long newOffset = halfOffset;
            long i = halfOffset;
            while (true) {
                if (++i > transactionCheckMax) {
                    break;
                }
                if (System.currentTimeMillis() - startTime > MAX_PROCESS_TIME_LIMIT) {
                    break;
                }
                MessageExt msgOp = getHalfMessage(opOffset);
                if (msgOp == null) {
                    newOffset = i;
                    break;
                }
                
                long valueOfCurrentHalfMsg = Long.parseLong(msgOp.getUserProperty(
                        MessageConst.PROPERTY_TRANSACTION_CHECK_TIMES));
                if (valueOfCurrentHalfMsg >= transactionCheckMax) {
                    listener.resolveHalfMsg(msgOp);
                } else {
                    listener.putBackHalfMsgQueue(msgOp, i);
                }
                newOffset = i;
            }
            if (newOffset != halfOffset) {
                transactionalMessageBridge.updateConsumeOffset(messageQueue, newOffset);
            }
        }
    } catch (Throwable e) {
        log.error("Check error", e);
    }
}

📖 学习资源

视频教程

• RocketMQ事务消息源码解析

书籍推荐

• 《RocketMQ技术内幕》 - 丁威
• 《分布式事务处理》 - 相关技术书籍

网站资源

• RocketMQ官方文档

✅ 学习检查清单

• [ ] 已理解事务消息流程
• [ ] 已理解状态回查机制
• [ ] 已理解事务消息存储
• [ ] 已理解事务消息消费
• [ ] 已绘制事务消息流程图
• [ ] 已总结核心知识点

---

上一章节：消息投递流程
下一章节：负载均衡机制
返回目录：RocketMQ源码深度解析