RocketMQ原理学习---生产者事物消息发送

Myth丶恋晨 2022-04-11 02:45 336阅读 0赞

上一篇博客[《RocketMQ原理学习---生产者普通消息发送》][RocketMQ_---]我们已经对生产者发送普通消息有了简单的了解，这篇博客我们来学习一下RocketMQ在发送事物消息时做了什么处理操作。

### 一、生产者发送消息 ###

RocketMQ通过实现2PC协议来实现分布式事物，RocketMQ事物消息发送与消费流程图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxOTI0ODYyMDc3_size_16_color_FFFFFF_t_70][]

接下来我们通过源码看看RocketMQ生产者在发送事物消息的时候做了什么操作。

调用事物消息接口发送事物消息。

public TransactionSendResult sendMessageInTransaction(Message msg, LocalTransactionExecuter tranExecuter, Object arg) throws MQClientException {
            if (null == this.transactionCheckListener) {
                throw new MQClientException("localTransactionBranchCheckListener is null", (Throwable)null);
            } else {
                //发送事物消息
                return this.defaultMQProducerImpl.sendMessageInTransaction(msg, tranExecuter, arg);
            }
        }

在sendMessageInTransaction可以看到会进行如下操作：

（1）调用 sendResult = this.send(msg); 发送消息获取结果

（2）执行本地事物localTransactionState = tranExecuter.executeLocalTransactionBranch(msg, arg);

（3）将执行本地事物的结果再调用this.endTransaction(sendResult, localTransactionState, localException);

public TransactionSendResult sendMessageInTransaction(Message msg, LocalTransactionExecuter tranExecuter, Object arg) throws MQClientException {
            if (null == tranExecuter) {
                throw new MQClientException("tranExecutor is null", (Throwable)null);
            } else {
                Validators.checkMessage(msg, this.defaultMQProducer);
                SendResult sendResult = null;
                MessageAccessor.putProperty(msg, "TRAN_MSG", "true");
                MessageAccessor.putProperty(msg, "PGROUP", this.defaultMQProducer.getProducerGroup());
    
                try {
                    //发送消息
                    sendResult = this.send(msg);
                } catch (Exception var10) {
                    throw new MQClientException("send message Exception", var10);
                }
    
                LocalTransactionState localTransactionState = LocalTransactionState.UNKNOW;
                Throwable localException = null;
                switch(sendResult.getSendStatus()) {
                case SEND_OK:
                    try {
                        if (sendResult.getTransactionId() != null) {
                            msg.putUserProperty("__transactionId__", sendResult.getTransactionId());
                        }
                        //执行本地事物
                        localTransactionState = tranExecuter.executeLocalTransactionBranch(msg, arg);
                        if (null == localTransactionState) {
                            localTransactionState = LocalTransactionState.UNKNOW;
                        }
    
                        if (localTransactionState != LocalTransactionState.COMMIT_MESSAGE) {
                            this.log.info("executeLocalTransactionBranch return {}", localTransactionState);
                            this.log.info(msg.toString());
                        }
                    } catch (Throwable var9) {
                        this.log.info("executeLocalTransactionBranch exception", var9);
                        this.log.info(msg.toString());
                        localException = var9;
                    }
                    break;
                case FLUSH_DISK_TIMEOUT:
                case FLUSH_SLAVE_TIMEOUT:
                case SLAVE_NOT_AVAILABLE:
                    localTransactionState = LocalTransactionState.ROLLBACK_MESSAGE;
                }
    
                try {
                    //提交执行结果
                    this.endTransaction(sendResult, localTransactionState, localException);
                } catch (Exception var8) {
                    this.log.warn("local transaction execute " + localTransactionState + ", but end broker transaction failed", var8);
                }
    
                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;
            }
        }

在endTransaction方法中会根据本地事物执行结果的状态，将状态信息发送到broker

public void endTransaction(SendResult sendResult, LocalTransactionState localTransactionState, Throwable localException) throws RemotingException, MQBrokerException, InterruptedException, UnknownHostException {
            MessageId id;
            if (sendResult.getOffsetMsgId() != null) {
                id = MessageDecoder.decodeMessageId(sendResult.getOffsetMsgId());
            } else {
                id = MessageDecoder.decodeMessageId(sendResult.getMsgId());
            }
    
            String transactionId = sendResult.getTransactionId();
            String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(sendResult.getMessageQueue().getBrokerName());
            EndTransactionRequestHeader requestHeader = new EndTransactionRequestHeader();
            requestHeader.setTransactionId(transactionId);
            requestHeader.setCommitLogOffset(id.getOffset());
            switch(localTransactionState) {
            case COMMIT_MESSAGE:
                requestHeader.setCommitOrRollback(8);
                break;
            case ROLLBACK_MESSAGE:
                requestHeader.setCommitOrRollback(12);
                break;
            case UNKNOW:
                requestHeader.setCommitOrRollback(0);
            }
    
            requestHeader.setProducerGroup(this.defaultMQProducer.getProducerGroup());
            requestHeader.setTranStateTableOffset(sendResult.getQueueOffset());
            requestHeader.setMsgId(sendResult.getMsgId());
            String remark = localException != null ? "executeLocalTransactionBranch exception: " + localException.toString() : null;
            this.mQClientFactory.getMQClientAPIImpl().endTransactionOneway(brokerAddr, requestHeader, remark, (long)this.defaultMQProducer.getSendMsgTimeout());
        }

### 二、Broker处理事物消息 ###

一条完整消息的要经过如下处理：

（1）将消息持久化到commitlog文件中

（2）创建consumequeue文件

（3）创建index文件

1、普通消息在消息发送成功后会进行上面三步创建操作

2、事物消息prepare阶段只会进行第一步commitlog文件创建（此时消费者是无法消费消息的），当commit阶段时会进行第二部操作创建consumequeue文件，这样消费者才可以消费消息。

class CommitLogDispatcherBuildConsumeQueue implements CommitLogDispatcher {
    
            @Override
            public void dispatch(DispatchRequest request) {
                final int tranType = MessageSysFlag.getTransactionValue(request.getSysFlag());
                switch (tranType) {
                    //普通消息
                    case MessageSysFlag.TRANSACTION_NOT_TYPE:
                    //确认事物消息
                    case MessageSysFlag.TRANSACTION_COMMIT_TYPE:
                        DefaultMessageStore.this.putMessagePositionInfo(request);
                        break;
                    //准备事物消息
                    case MessageSysFlag.TRANSACTION_PREPARED_TYPE:
                    //回滚事物消息
                    case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE:
                        break;
                }
            }
        }

[RocketMQ_---]: https://blog.csdn.net/qq924862077/article/details/83928643
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxOTI0ODYyMDc3_size_16_color_FFFFFF_t_70]: /images/20220411/1706299b6a924027a48b222c546bff83.png