RocketMQ事务消息如何实现-创新互联

这篇文章主要介绍了RocketMQ事务消息如何实现,具有一定借鉴价值,感兴趣的朋友可以参考下,希望大家阅读完这篇文章之后大有收获,下面让小编带着大家一起了解一下。

站在用户的角度思考问题,与客户深入沟通,找到瀍河网站设计与瀍河网站推广的解决方案,凭借多年的经验,让设计与互联网技术结合,创造个性化、用户体验好的作品,建站类型包括:成都做网站、网站制作、成都外贸网站建设、企业官网、英文网站、手机端网站、网站推广、主机域名、虚拟空间、企业邮箱。业务覆盖瀍河地区。

根据上文的描述,发送事务消息的入口为:

TransactionMQProducer#sendMessageInTransaction:public TransactionSendResult sendMessageInTransaction(final Message msg, final Object arg) throws MQClientException {        if (null == this.transactionListener) {    // @1
            throw new MQClientException("TransactionListener is null", null);
        }        return this.defaultMQProducerImpl.sendMessageInTransaction(msg, transactionListener, arg);  // @2
    }

代码@1:如果transactionListener为空,则直接抛出异常。
代码@2:调用defaultMQProducerImpl的sendMessageInTransaction方法。

接下来重点分享sendMessageInTransaction方法

DefaultMQProducerImpl#sendMessageInTransactionpublic TransactionSendResult sendMessageInTransaction(final Message msg,           final TransactionListener tranExecuter, final Object arg)  throws MQClientException {

Step1:首先先阐述一下参数含义。

  • final Message msg:消息

  • TransactionListener tranExecuter:事务监听器

  • Object arg:其他附加参数

DefaultMQProducerImpl#sendMessageInTransaction
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);
}

Step2:在消息属性中增加两个属性:TRAN_MSG,其值为true,表示为事务消息;PGROUP:消息所属发送者组,然后以同步方式发送消息。在消息发送之前,会先检查消息的属性TRAN_MSG,如果存在并且值为true,则通过设置消息系统标记的方式,设置消息为MessageSysFlag.TRANSACTION_PREPARED_TYPE。

DefaultMQProducerImpl#sendKernelImplfinal String tranMsg = msg.getProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED);if (tranMsg != null && Boolean.parseBoolean(tranMsg)) {
       sysFlag |= MessageSysFlag.TRANSACTION_PREPARED_TYPE;
}
SendMessageProcessor#sendMessage
String traFlag = oriProps.get(MessageConst.PROPERTY_TRANSACTION_PREPARED);if (traFlag != null && Boolean.parseBoolean(traFlag)) {        if (this.brokerController.getBrokerConfig().isRejectTransactionMessage()) {
             response.setCode(ResponseCode.NO_PERMISSION);
             response.setRemark(                    "the broker[" + this.brokerController.getBrokerConfig().getBrokerIP1()
                        + "] sending transaction message is forbidden");             return response;
       }
      putMessageResult = this.brokerController.getTransactionalMessageService().prepareMessage(msgInner);
} else {
      putMessageResult = this.brokerController.getMessageStore().putMessage(msgInner);
}

Step3:Broker端收到客户端发送消息请求后,判断消息类型。如果是事务消息,则调用TransactionalMessageService#prepareMessage方法,否则走原先的逻辑,调用MessageStore#putMessage方法将消息存入Broker服务端。
本节重点阐述事务消息的实现原理,故接下来将重点关注prepareMessage方法,如想了解RocketMQ消息存储相关,可以关注作者 源码分析RocketMQ系列 。

org.apache.rocketmq.broker.transaction.queue.TransactionalMessageServiceImpl#prepareMessagepublic PutMessageResult prepareMessage(MessageExtBrokerInner messageInner) {        return transactionalMessageBridge.putHalfMessage(messageInner);
 }

step4:事务消息,将调用TransactionalMessageServiceImpl#prepareMessage方法,继而调用TransactionalMessageBridge#prepareMessage方法。

TransactionalMessageBridge#parseHalfMessageInnerpublic PutMessageResult putHalfMessage(MessageExtBrokerInner messageInner) {        return store.putMessage(parseHalfMessageInner(messageInner));
    }    private MessageExtBrokerInner parseHalfMessageInner(MessageExtBrokerInner msgInner) {
        MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_REAL_TOPIC, msgInner.getTopic());
        MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_REAL_QUEUE_ID,
            String.valueOf(msgInner.getQueueId()));
        msgInner.setSysFlag(
            MessageSysFlag.resetTransactionValue(msgInner.getSysFlag(), MessageSysFlag.TRANSACTION_NOT_TYPE));
        msgInner.setTopic(TransactionalMessageUtil.buildHalfTopic());
        msgInner.setQueueId(0);
        msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties()));        return msgInner;
    }

Step5:备份消息的原主题名称与原队列ID,然后取消消息的事务消息标签,重新设置消息的主题为:RMQ_SYS_TRANS_HALF_TOPIC,队列ID固定为0。然后调用MessageStore#putMessage方法将消息持久化,这里TransactionalMessageBridge桥接类,就是封装事务消息的相关流程,最终调用MessageStore完成消息的持久化。消息入库后,会继续回到DefaultMQProducerImpl#sendMessageInTransaction,上文的Step2后面,也就是通过同步将消息发送到消息服务端。

注:这是事务消息Prepare状态的处理逻辑,消息是存储在消息服务器了,但存储的并不是原主题,而是RMQ_SYS_TRANS_HALF_TOPIC,故此时消费端是无法消费shen
生产者发送的消息的。看到这里,如果对RocketMQ比较熟悉的话,肯定会有一个“定时任务”去取这个主题下的消息,然后则“合适”的时机将消息的主题恢复。

DefaultMQProducerImpl#sendMessageInTransactionswitch (sendResult.getSendStatus()) {            case SEND_OK: {                try {                    if (sendResult.getTransactionId() != null) {
                        msg.putUserProperty("__transactionId__", sendResult.getTransactionId());
                    }
                    String transactionId = msg.getProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX);                    if (null != transactionId && !"".equals(transactionId)) {
                        msg.setTransactionId(transactionId);
                    }
                    localTransactionState = tranExecuter.executeLocalTransaction(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;
        }

Step6:如果消息发送成功,会回调TransactionListener#executeLocalTransaction方法,执行本地事务,并且返回本地事务状态为LocalTransactionState,枚举值如下:

  • COMMIT_MESSAGE,

  • ROLLBACK_MESSAGE,

  • UNKNOW

注意:TransactionListener#executeLocalTransaction是在发送者成功发送PREPARED消息后,会执行本地事务方法,然后返回本地事务状态;如果PREPARED消息发送失败,则不会调用TransactionListener#executeLocalTransaction,并且本地事务消息,设置为LocalTransactionState.ROLLBACK_MESSAGE,表示消息需要被回滚。

DefaultMQProducerImpl#sendMessageInTransactiontry {this.endTransaction(sendResult, localTransactionState, localException);
} catch (Exception e) {
log.warn("local transaction execute " + localTransactionState + ", but end broker transaction failed", e);
}

step7:调用endTransaction方法结束事务(提交或回滚)。

DefaultMQProducerImpl#endTransaction
EndTransactionRequestHeader requestHeader = new EndTransactionRequestHeader();
requestHeader.setTransactionId(transactionId);
requestHeader.setCommitLogOffset(id.getOffset());switch (localTransactionState) {    case COMMIT_MESSAGE:
         requestHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_COMMIT_TYPE);         break;    case ROLLBACK_MESSAGE:
         requestHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_ROLLBACK_TYPE);         break;     case UNKNOW:
         requestHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_NOT_TYPE);         break;     default:         break;
}
requestHeader.setProducerGroup(this.defaultMQProducer.getProducerGroup());
requestHeader.setTranStateTableOffset(sendResult.getQueueOffset());
requestHeader.setMsgId(sendResult.getMsgId());

step8:组装结束事务请求,主要参数为:事务ID、事务操作(commitOrRollback)、消费组、消息队列偏移量、消息ID,fromTransactionCheck,从这里发出的请求,默认为false。Broker端的请求处理器为:EndTransactionProcessor。

step9:EndTransactionProcessor根据事务提交类型:TRANSACTION_COMMIT_TYPE(提交事务)、TRANSACTION_ROLLBACK_TYPE(回滚事务)、TRANSACTION_NOT_TYPE(忽略该请求)。

到目前为止,已详细梳理了RocketMQ事务消息的发送流程,更加准确的说是Prepare状态的消息发送流程。具体流程如图所示:

本文到这里,初步展示了事务消息的发送流程,总的说来,RocketMQ的事务消息发送使用二阶段提交思路,首先,在消息发送时,先发送消息类型为Prepread类型的消息,然后在将该消息成功存入到消息服务器后,会回调TransactionListener#executeLocalTransaction,执行本地事务状态回调函数,然后根据该方法的返回值,结束事务:
1、COMMIT_MESSAGE :提交事务。
2、ROLLBACK_MESSAGE:回滚事务。
3、UNKNOW:未知事务状态,此时消息服务器(Broker)收到EndTransaction命令时,将不对这种消息做处理,消息还处于Prepared类型,存储在主题为:RMQ_SYS_TRANS_HALF_TOPIC的队列中,然后消息发送流程将结束,那这些消息如何提交或回滚呢?

为了实现避免客户端需要再次发送提交、回滚命令,RocketMQ会采取定时任务将RMQ_SYS_TRANS_HALF_TOPIC中的消息取出,然后回到客户端,判断该消息是否需要提交或回滚,来完成事务消息的声明周期。

感谢你能够认真阅读完这篇文章,希望小编分享的“RocketMQ事务消息如何实现”这篇文章对大家有帮助,同时也希望大家多多支持创新互联,关注创新互联-成都网站建设公司行业资讯频道,更多相关知识等着你来学习!


分享文章:RocketMQ事务消息如何实现-创新互联
网页URL:http://pwwzsj.com/article/dpjhgs.html