RocketMQ原理学习---生产者事物消息发送-蒲公英云

    上一篇博客[《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;
            }
        }
    }