RocketMQ源码 — 十一、 RocketMQ事务消息

分布式事务是一个复杂的问题,rmq实现了事务的最终一致性,rmq保证本地事务成功消息一定会发送成功并被成功消费,如果本地事务失败了,消息不会被发送。rmq事务消息的实现过程为:
  • producer发送half消息
  • broker确认half消息,并通知producer,表示消息已经成功发送到broker(这个过程其实就是步骤1broker的返回)
  • producer收到half确认消息之后,执行自己本地事务,并将事务结果(UNKNOW、commit、rollback)告诉broker(这是一个oneway消息,而且失败不重试)
  • broker收到producer本地事务的结果后决定是否投递消息给consumer
  • 鉴于producer发送本地事务结果可能失败,broker会定时扫描集群中的事务消息,然后回查(apache4.2.0尚未实现,因为没有调用org.apache.rocketmq.broker.client.net.Broker2Client#checkProducerTransactionState)
  • producer发送half消息事务消息的发送过程和普通消息发送过程是不一样的,发送消息的方法是org.apache.rocketmq.client.producer.TransactionMQProducer#sendMessageInTransaction,入参有一个LocalTransactionExecuter,需要用户实现一个本地事务的executor,用户可以在executor中执行事务操作
    // org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl#sendMessageInTransaction
    public TransactionSendResult sendMessageInTransaction(final Message msg,
        final LocalTransactionExecuter tranExecuter, final Object arg)
        throws MQClientException {
        if (null == tranExecuter) {
            throw new MQClientException("tranExecutor is null", null);
        }
        Validators.checkMessage(msg, this.defaultMQProducer);
    
        SendResult sendResult = null;
        // 标记消息是half消息
        MessageAccessor.putProperty(msg, MessageConst.PROPERTY_TRANSACTION_PREPARED, "true");
        MessageAccessor.putProperty(msg, MessageConst.PROPERTY_PRODUCER_GROUP, this.defaultMQProducer.getProducerGroup());
        try {
            // 发送half消息,该方法是同步发送,事务消息也必须是同步发送
            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: {
                // 只有在half消息发送成功的时候才会执行事务
                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) {
                        log.info("executeLocalTransactionBranch return {}", localTransactionState);
                        log.info(msg.toString());
                    }
                } catch (Throwable e) {
                    log.info("executeLocalTransactionBranch exception", e);
                    log.info(msg.toString());
                    localException = e;
                }
            }
            break;
            case FLUSH_DISK_TIMEOUT:
            case FLUSH_SLAVE_TIMEOUT:
            case SLAVE_NOT_AVAILABLE:
                localTransactionState = LocalTransactionState.ROLLBACK_MESSAGE;
                break;
            default:
                break;
        }
    
        try {
            // 根据事务commit的情况来判断下一步操作
            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;
    }
    为了保证本地事务和消息发送成功的原子性,producer会先发送一个half消息到broker
  • 只有half消息发送成功了,事务才会被执行
  • 如果half消息发送失败了,事务不会被执行
  • half消息和普通的消息也不一样,half消息发送到broker后并不会被consumer消费掉。之所以不会被消费掉的原因如下:
  • broker在将消息写入CommitLog的时候会判断消息类型,如果是是prepare或者rollback消息,ConsumeQueue的offset(每个消息对应ConsumeQueue中的一个数据结构(包含topic、tag的hashCode、消息对应CommitLog的物理offset),offset表示数据结构是第几个)不会增加
  • broker在构造ConsumeQueue的时候会判断是否是prepare或者rollback消息,如果是这两种中的一种则不会将该消息放入ConsumeQueue,cnosumer在拉取消息的时候也就不会拉取到prepare和rollback的消息。
  • 相关代码如下:
    // org.apache.rocketmq.store.CommitLog.DefaultAppendMessageCallback#doAppend(long, java.nio.ByteBuffer, int, org.apache.rocketmq.store.MessageExtBrokerInner)
    switch (tranType) {
        case MessageSysFlag.TRANSACTION_PREPARED_TYPE:
        case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE:
            break;
        case MessageSysFlag.TRANSACTION_NOT_TYPE:
        case MessageSysFlag.TRANSACTION_COMMIT_TYPE:
            // The next update ConsumeQueue information
            CommitLog.this.topicQueueTable.put(key, ++queueOffset);
            break;
        default:
            break;
    }
    
    // org.apache.rocketmq.store.DefaultMessageStore.CommitLogDispatcherBuildConsumeQueue#dispatch
    
    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;
        }
    }
    以上两点保证了prepare消息也就是half消息不会被消费。producer结束事务producer根据half消息发送结果和事务执行结果来处理事务——commit或者rollback。从上面发送消息的代码可以看到最后调用了endTransaction来处理事务执行结果,这个方法里面就是将事务执行的结果通过消息发送给broker,由broker决定消息是否投递。
    public void endTransaction(
        final SendResult sendResult,
        final LocalTransactionState localTransactionState,
        final Throwable localException) throws RemotingException, MQBrokerException, InterruptedException, UnknownHostException {
        final MessageId id;
        // 从broker返回的信息中获取half消息的offset
        if (sendResult.getOffsetMsgId() != null) {
            id = MessageDecoder.decodeMessageId(sendResult.getOffsetMsgId());
        } else {
            id = MessageDecoder.decodeMessageId(sendResult.getMsgId());
        }
        String transactionId = sendResult.getTransactionId();
        final String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(sendResult.getMessageQueue().getBrokerName());
        EndTransactionRequestHeader requestHeader = new EndTransactionRequestHeader();
        // 需要把transactionId和offset发送给broker,便于broker查找half消息
        requestHeader.setTransactionId(transactionId);
        requestHeader.setCommitLogOffset(id.getOffset());
        switch (localTransactionState) {
            case COMMIT_MESSAGE:
                // 表明本地址事务成功commit,告诉broker可以提交事务
                requestHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_COMMIT_TYPE);
                break;
            case ROLLBACK_MESSAGE:
                // 说明事物需要回滚,有可能是half消息发送失败,也有可能是本地事务执行失败
                requestHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_ROLLBACK_TYPE);
                break;
            case UNKNOW:
                // 如果状态是UNKNOW,broker还会反查producer,也就是接口:org.apache.rocketmq.example.transaction.TransactionCheckListenerImpl#checkLocalTransactionState的作用,但是目前rmq4.2.0并没有向producer查询,也就是源码中都没有调用这个接口
                requestHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_NOT_TYPE);
                break;
            default:
                break;
        }
    
        requestHeader.setProducerGroup(this.defaultMQProducer.getProducerGroup());
        requestHeader.setTranStateTableOffset(sendResult.getQueueOffset());
        requestHeader.setMsgId(sendResult.getMsgId());
        String remark = localException != null ? ("executeLocalTransactionBranch exception: " + localException.toString()) : null;
        // 这个发送消息是onway的,也就是不会等待返回
        this.mQClientFactory.getMQClientAPIImpl().endTransactionOneway(brokerAddr, requestHeader, remark,
            this.defaultMQProducer.getSendMsgTimeout());
    }
    从最一开始的图看出,producer将事务结果的消息发送给broker的时候可能会失败,如果失败了,broker就不知道本次事务是否应该commit,为了防止这种情况,rmq会向producer发送一个command查询处于prepare状态的事务的结果,上面也说了rmq4.2.0并没有发送这个command,也就是说当前rmq并不能保证producer将事务结果通知到broker。broker决定消息是否可以投递broker处理事务结果的消息的类是org.apache.rocketmq.broker.processor.EndTransactionProcessor
  • 收到消息之后先检查是否是事务类型的消息,不是事务消息直接返回。
  • 根据header中的offset查询half消息,查不到直接返回,不作处理
  • 根据half消息构造新的消息,新构造的这个消息会被重新写入CommitLog,如果是rollback消息则body为空
  • 如果是rollback消息的话,该消息不会被投递(原因和half不会被投递的原因一样),commit消息broker才会投递给consumer
  • 也就是说rmq对于commit和rollback都会新写一个消息到CommitLog,只是rollback的消息的body是空的,而且该消息和half消息一样不会被投递,直到CommitLog删除过期消息,会从磁盘中删除;但是commit的时候,rmq会重新封装half消息并“投递”给consumer消费。consumer保证消费成功关于事务消息consumer端的消费方式和普通消息是一样的,RocketMQ能保证消息能被consumer收到(消息重试等机制,其实有可能存在consumer消费失败的情况,这种情况RocketMQ并不能解决,官方建议人工解决,这种情况出现的概率极低)。总结基于rmq的阿里云ons实现了事务最终一致性的所有功能,但是apache rmq没有实现消息回查的功能。所以rmq存在一定几率会让事务处于事务结果不明确的状态。参考收发事务消息

    相关内容推荐