事务信息是RocketMQ的一个十分特征的初级个性，它的基础诉求是经过RocketMQ的事务机制，来保证高低游的数据⼀致性。

咱们在单机版本上方只有要在业务方法上加上对应的事务就可以到达成果，然而散布式的场景下，多个系统之间的协分配合，你无法知道究竟是那个先口头那个后口头，当然在微服务外面存在Seate框架来保证事务，然而这事务的保证一直是心头大患，只能用一句话描画鱼和熊掌无法兼得。

而RocketMq的事务信息能够在优化性能的状况下满足要求，其关键成功是支持散布式状况下保证信息消费和本地事务的最终分歧性，信息消费咱们可以经常使用顺序信息去口头，这样咱们只有要满足这两个的事务即可。

成功环节

图片

预备阶段：消费者将信息发送到Broker，Broker向消费者发送ack示意信息发送成功，然而此时的信息为一个期待形态，不会被消费者去消费。（消费者继续口头接上去的代码）

确认阶段：当咱们口头完一切的代码后，本地事务要么回滚要么提交，此时当咱们了解本地事务的形态后，将结果推送给Broker做二次确认结果，假设为Commit则将修正激活预备推送给消费者，假设为Rollback则将信息启动回滚。

补救机制：当出现意外状况没有出现二次确认，此时咱们在固定期间后将会启动回查，审核回查信息对应的本地事务的形态，重写Commit或许Rollback。

触及形态以及留意点

事务信息存在三种形态：

CommitTransaction：提交事务形态，此形态下准许消费者消费。

RollbackTransaction：回滚事务形态，此形态下信息会被删除。

Unknown：两边形态，此形态下会期待本地事务处置结果启动对应操作。

留意点：

本信息形态是一种抵消费者无法见的形态，将信息的内容放到系统Topic的RMQ_SYS_TRANS_HALF_TOPIC队列外面去。

事务信息中的相关参数可以启动设置，比如：本地事务回查次数transactionCheckMax自动15次，本地事务回查的间隙transactionCheckInterval自动60s，超出后会间接将信息摈弃。

RocketMQ的事务信息是指运行本地事务和发送信息操作可以定义到全局事务中，要么同时成功，要么同时失败，经过RocketMQ的事务信息可以成功牢靠信息的最终分歧性打算。

源码解析

Producer端经过构建TransactionMQProducer对象绑定事务监听。

TransactionListenertransactionListener=newTransactionListener(){@OverridepublicLocalTransactionStateexecuteLocalTransaction(Messagemsg,Objectarg){returnLocalTransactionState.COMMIT_MESSAGE;}@OverridepublicLocalTransactionStatecheckLocalTransaction(MessageExtmsg){returnLocalTransactionState.COMMIT_MESSAGE;}};TransactionMQProducerproducer=newTransactionMQProducer(producerGroupTemp);producer.setTransactionListener(transactionListener);producer.setNamesrvAddr("127.0.0.1:9876");product.start();SendResultresult=producer.sendMessageInTransaction(message,arg);

口头sendMessageInTransaction方法来发送信息。

publicTransactionSendResultsendMessageInTransaction(finalMessagemsg,finalLocalTransactionExecuterlocalTransactionExecuter,finalObjectarg)throwsMQClientException{//审核TransactionListener能否存在，假设不存在就间接抛意外TransactionListenertransactionListener=getCheckListener();if(null==localTransactionExecuter&&null==transactionListener){thrownewMQClientException("tranExecutorisnull",null);}//事务信息不支持提前等个性if(msg.getDelayTimeLevel()!=0){MessageAccessor.clearProperty(msg,MessageConst.PROPERTY_DELAY_TIME_LEVEL);}Validators.checkMessage(msg,this.defaultMQProducer);SendResultsendResult=null;//设置half属性，标明是事务属性MessageAccessor.putProperty(msg,MessageConst.PROPERTY_TRANSACTION_PREPARED,"true");//设置所属生成者组//broker向消费者发送回查事务恳求依据这个producergroup找到指定的channel//消费者能找到一切在同一个组的机器实例从而审核事务形态MessageAccessor.putProperty(msg,MessageConst.PROPERTY_PRODUCER_GROUP,this.defaultMQProducer.getProducerGroup());try{//同步发送sendResult=this.send(msg);}catch(Exceptione){thrownewMQClientException("sendmessageException",e);}LocalTransactionStatelocalTransactionState=LocalTransactionState.UNKNOW;ThrowablelocalException=null;//信息前往信息switch(sendResult.getSendStatus()){//第一阶段信息发送成功caseSEND_OK:{try{if(sendResult.getTransactionId()!=null){//设置事务ID属性msg.putUserProperty("__transactionId__",sendResult.getTransactionId());}StringtransactionId=msg.getProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX);if(null!=transactionId&&!"".equals(transactionId)){msg.setTransactionId(transactionId);}if(null!=localTransactionExecuter){//口头本地事务localTransactionState=localTransactionExecuter.executeLocalTransactionBranch(msg,arg);}elseif(transactionListener!=null){log.debug("UsednewtransactionAPI");//发送信息成功后，口头本地操作localTransactionState=transactionListener.executeLocalTransaction(msg,arg);}if(null==localTransactionState){localTransactionState=LocalTransactionState.UNKNOW;}if(localTransactionState!=LocalTransactionState.COMMIT_MESSAGE){log.info("executeLocalTransactionBranchreturn{}",localTransactionState);log.info(msg.toString());}}catch(Throwablee){log.info("executeLocalTransactionBranchexception",e);log.info(msg.toString());localException=e;}}break;caseFLUSH_DISK_TIMEOUT:caseFLUSH_SLAVE_TIMEOUT:caseSLAVE_NOT_AVLABLE:localTransactionState=LocalTransactionState.ROLLBACK_MESSAGE;break;default:break;}try{//本地事务口头终了向broker提交事务或回滚事务this.endTransaction(msg,sendResult,localTransactionState,localException);}catch(Exceptione){log.warn("localtransactionexecute"+localTransactionState+",butendbrokertransactionfailed",e);}TransactionSendResulttransactionSendResult=newTransactionSendResult();transactionSendResult.setSendStatus(sendResult.getSendStatus());transactionSendResult.setMessageQueue(sendResult.getMessageQueue());transactionSendResult.setMsgId(sendResult.getMsgId());transactionSendResult.setQueueOffset(sendResult.getQueueOffset());transactionSendResult.setTransactionId(sendResult.getTransactionId());transactionSendResult.setLocalTransactionState(localTransactionState);returntransactionSendResult;}

首先发送第一阶段信息间接前往半提交形态，而后口头本地事务前往事务的三种形态，未知，回滚，提交，最后口头endTransaction方法，把事务口头的形态通知broker。

endTransaction方法

依据本地事务口头形态构建requestHeader对象口头二阶段提交。

publicvoidendTransaction(finalMessagemsg,finalSendResultsendResult,finalLocalTransactionStatelocalTransactionState,finalThrowablelocalException)throwsRemotingException,MQBrokerException,InterruptedException,UnknownHostException{finalMessageIdid;//失掉信息中的MessageIdif(sendResult.getOffsetMsgId()!=null){id=MessageDecoder.decodeMessageId(sendResult.getOffsetMsgId());}else{id=MessageDecoder.decodeMessageId(sendResult.getMsgId());}StringtransactionId=sendResult.getTransactionId();//找到broker地址finalStringbrokerAddr=this.mQClientFactory.findBrokerAddressInPublish(sendResult.getMessageQueue().getBrokerName());//构建EndTransactionRequestHeader对象EndTransactionRequestHeaderrequestHeader=newEndTransactionRequestHeader();requestHeader.setTransactionId(transactionId);//offset是prepare信息中offsetMsgId中失掉的requestHeader.setCommitLogOffset(id.getOffset());requestHeader.setBname(sendResult.getMessageQueue().getBrokerName());//社会提交/回滚形态switch(localTransactionState){caseCOMMIT_MESSAGE://提交requestHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_COMMIT_TYPE);break;caseROLLBACK_MESSAGE://回滚requestHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_ROLLBACK_TYPE);break;caseUNKNOW://未知requestHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_NOT_TYPE);break;default:break;}doExecuteEndTransactionHook(msg,sendResult.getMsgId(),brokerAddr,localTransactionState,false);requestHeader.setProducerGroup(this.defaultMQProducer.getProducerGroup());requestHeader.setTranStateTableOffset(sendResult.getQueueOffset());requestHeader.setMsgId(sendResult.getMsgId());Stringremark=localException!=null?("executeLocalTransactionBranchexception:"+localException.toString()):null;//发送给broker端this.mQClientFactory.getMQClientAPIImpl().endTransactionOneway(brokerAddr,requestHeader,remark,this.defaultMQProducer.getSendMsgTimeout());}

将本地方法口头事务的结果发送给Broker，经过endTransactionOneway方法创立Code为END_TRANSACTION的信息，而后在Broker就会找出对应的Processor来处置。

Broker端处置

Broker总共存在两个处置，首先针对第一个阶段发送的Half信息，broker要启动相关的操作，前面endTransaction提交出去的事务形态，针对三种形态启动相关操作。

接纳第一阶段发送的Half信息

SendMessageProcessor的sendMessage方法中去口头处置事务信息。

//发送Half信息时，在属性中设置了PROPERTY_TRANSACTION_PREPARED为true，这里依据这个属性判别能否是事务信息StringtraFlag=oriProps.get(MessageConst.PROPERTY_TRANSACTION_PREPARED);if(Boolean.parseBoolean(traFlag)&&!(msgInner.getReconsumeTimes()>0&&msgInner.getDelayTimeLevel()>0)){//Forclientunderversion4.6.1if(this.brokerController.getBrokerConfig().isRejectTransactionMessage()){response.setCode(ResponseCode.NO_PERMISSION);response.setRemark("thebroker["+this.brokerController.getBrokerConfig().getBrokerIP1()+"]sendingtransactionmessageisforbidden");returnresponse;}//事务信息进入这里，把信息的topic改成RMQ_SYS_TRANS_HALF_TOPIC，以同步刷盘的形式存入storeputMessageResult=this.brokerController.getTransactionalMessageService().prepareMessage(msgInner);}

假设信息携带事务标志就去口头TransactionMessageService类的prepareMessage方法启动相关的处置。

//解析Half信息privateMessageExtBrokerInnerparseHalfMessageInner(MessageExtBrokerInnermsgInner){//把实在的topic和实在的queueId放在信息的属性中MessageAccessor.putProperty(msgInner,MessageConst.PROPERTY_REAL_TOPIC,msgInner.getTopic());MessageAccessor.putProperty(msgInner,MessageConst.PROPERTY_REAL_QUEUE_ID,String.valueOf(msgInner.getQueueId()));//设置自动的事务形态为TRANSACTION_NOT_TYPE=>unknowmsgInner.setSysFlag(MessageSysFlag.resetTransactionValue(msgInner.getSysFlag(),MessageSysFlag.TRANSACTION_NOT_TYPE));//将信息的topic设置为RMQ_SYS_TRANS_HALF_TOPIC，这个是抵消费者无法见的msgInner.setTopic(TransactionalMessageUtil.buildHalfTopic());//设置queueId=0msgInner.setQueueId(0);msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties()));returnmsgInner;}

启动topic的切换，将原来的topic存入到信息的属性外面，将信息的topic设置为RMQ_SYS_TRANS_HALF_TOPIC。

处置endTransaction方法

在endTransaction方法中将信息同步给Broker处置的Code对应为END_TRANSACTION，Broker就会找出对应的Processor来处置该类即调用EndTransactionProcessor类的processRequest方法处置。

if(MessageSysFlag.TRANSACTION_COMMIT_TYPE==requestHeader.getCommitOrRollback()){//依据commitLogOffset失掉文件中的message，失掉到了前往successresult=this.brokerController.getTransactionalMessageService().commitMessage(requestHeader);if(result.getResponseCode()==ResponseCode.SUCCESS){//审核信息能否分歧RemotingCommandres=checkPrepareMessage(result.getPrepareMessage(),requestHeader);if(res.getCode()==ResponseCode.SUCCESS){//生成要保留的信息MessageExtBrokerInnermsgInner=endMessageTransaction(result.getPrepareMessage());msgInner.setSysFlag(MessageSysFlag.resetTransactionValue(msgInner.getSysFlag(),requestHeader.getCommitOrRollback()));msgInner.setQueueOffset(requestHeader.getTranStateTableOffset());msgInner.setPreparedTransactionOffset(requestHeader.getCommitLogOffset());msgInner.setStoreTimestamp(result.getPrepareMessage().getStoreTimestamp());MessageAccessor.clearProperty(msgInner,MessageConst.PROPERTY_TRANSACTION_PREPARED);//把实在的topic信息存储到CommitLog中RemotingCommandsendResult=sendFinalMessage(msgInner);if(sendResult.getCode()==ResponseCode.SUCCESS){//移除prepare信息，存入opQueueMap中this.brokerController.getTransactionalMessageService().deletePrepareMessage(result.getPrepareMessage());}returnsendResult;}returnres;}//回滚}elseif(MessageSysFlag.TRANSACTION_ROLLBACK_TYPE==requestHeader.getCommitOrRollback()){//查问到half信息则前往成功result=this.brokerController.getTransactionalMessageService().rollbackMessage(requestHeader);if(result.getResponseCode()==ResponseCode.SUCCESS){//审核信息能否分歧RemotingCommandres=checkPrepareMessage(result.getPrepareMessage(),requestHeader);if(res.getCode()==ResponseCode.SUCCESS){//移除prepare信息，存入opQueueMap中this.brokerController.getTransactionalMessageService().deletePrepareMessage(result.getPrepareMessage());}returnres;}}

仅仅展现相关外围代码，其关键逻辑：首先去判别恳求的形式是commit还是rollback，假设是commit查问到信息恢复信息原来的topic，而后删除halftopic上的信息转存到opQueueMap中，假设是rollback间接启动删除halftopic上的信息并转存到opQueueMap中去。

留意：opQueueMap的引入为了处置有或许出现网络、进程、线程等各种要素造成消费端未能成功处置信息的状况，该机制的作用是在消费者端将未成功处置的信息从新发送到服务端启动重试，直到确认信息曾经被成功处置或许到达最大重试次数后启动回滚操作。而Op信息自身则是经过修正信息形态来成功的。

信息回查

当网络终止或许照应超时等各种意外信息造成信息并没有传送到broker端去，为了处置这一疑问在Broker就开启一个回查线程每隔一分钟口头一次性处置超越6s未回查的信息，当超越15次回查后间接将信息摈弃。

在启动BrokerController类时，会去调用startProcessorByHa方法假设是Master节点就会去启动一个线程每隔6s处置未回查的信息，审核最大次数为15次。

publicvoidrun(){log.info("Starttransactioncheckservicethread!");longcheckInterval=brokerController.getBrokerConfig().getTransactionCheckInterval();while(!this.isStopped()){this.waitForRunning(checkInterval);}log.info("Endtransactioncheckservicethread!");}protectedvoidonWaitEnd(){longtimeout=brokerController.getBrokerConfig().getTransactionTimeOut();intcheckMax=brokerController.getBrokerConfig().getTransactionCheckMax();longbegin=System.currentTimeMillis();log.info("Begintocheckpreparemessage,begintime:{}",begin);//审核回查信息timeout=6scheckMax=15this.brokerController.getTransactionalMessageService().check(timeout,checkMax,this.brokerController.getTransactionalMessageCheckListener());log.info("Endtocheckpreparemessage,consumedtime:{}",System.currentTimeMillis()-begin);}

在check方法外面去调用listener.resolveHalfMsg(msgExt)方法去处置事务信息。

publicvoidresolveHalfMsg(finalMessageExtmsgExt){executorService.execute(newRunnable(){@Overridepublicvoidrun(){try{sendCheckMessage(msgExt);}catch(Exceptione){LOGGER.error("Sendcheckmessageerror!",e);}}});}

口头sendCheckMessage方法发送一个审核事务形态的Code为CHECK_TRANSACTION_STATE的信息，在客户端MQClientAPIImpl初始化的时刻就会去注册一个Code对应的Processor，最终就会去口头checkTransactionState方法，判别本地事务的形态，而后再去口头endTransactionOneway动员END_TRANSACTION处置。

publicvoidcheckTransactionState(finalStringaddr,finalMessageExtmsg,finalCheckTransactionStateRequestHeaderheader){Runnablerequest=newRunnable(){privatefinalStringbrokerAddr=addr;privatefinalMessageExtmessage=msg;privatefinalCheckTransactionStateRequestHeadercheckRequestHeader=header;privatefinalStringgroup=DefaultMQProducerImpl.this.defaultMQProducer.getProducerGroup();//口头线程方法@Overridepublicvoidrun(){TransactionCheckListenertransactionCheckListener=DefaultMQProducerImpl.this.checkListener();TransactionListenertransactionListener=getCheckListener();if(transactionCheckListener!=null||transactionListener!=null){LocalTransactionStatelocalTransactionState=LocalTransactionState.UNKNOW;Throwableexception=null;try{if(transactionCheckListener!=null){localTransactionState=transactionCheckListener.checkLocalTransactionState(message);}elseif(transactionListener!=null){log.debug("UsednewcheckAPIintransactionmessage");//审核本地事务localTransactionState=transactionListener.checkLocalTransaction(message);}else{log.warn("CheckTransactionState,picktransactionListenerbygroup[{}]failed",group);}}catch(Throwablee){log.error("BrokercallcheckTransactionState,butcheckLocalTransactionStateexception",e);exception=e;}//处置事务形态this.processTransactionState(localTransactionState,group,exception);}else{log.warn("CheckTransactionState,picktransactionCheckListenerbygroup[{}]failed",group);}}//privatevoidprocessTransactionState(finalLocalTransactionStatelocalTransactionState,finalStringproducerGroup,finalThrowableexception){finalEndTransactionRequestHeaderthisHeader=newEndTransactionRequestHeader();thisHeader.setCommitLogOffset(checkRequestHeader.getCommitLogOffset());thisHeader.setProducerGroup(producerGroup);thisHeader.setTranStateTableOffset(checkRequestHeader.getTranStateTableOffset());thisHeader.setFromTransactionCheck(true);thisHeader.setBname(checkRequestHeader.getBname());StringuniqueKey=message.getProperties().get(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX);if(uniqueKey==null){uniqueKey=message.getMsgId();}thisHeader.setMsgId(uniqueKey);thisHeader.setTransactionId(checkRequestHeader.getTransactionId());switch(localTransactionState){//提交形态caseCOMMIT_MESSAGE:thisHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_COMMIT_TYPE);break;//回滚形态caseROLLBACK_MESSAGE:thisHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_ROLLBACK_TYPE);log.warn("whenbrokercheck,clientrollbackthistransaction,{}",thisHeader);break;//未知形态caseUNKNOW:thisHeader.setCommitOrRollback(MessageSysFlag.TRANSACTION_NOT_TYPE);log.warn("whenbrokercheck,clientdoesnotknowthistransactionstate,{}",thisHeader);break;default:break;}Stringremark=null;if(exception!=null){remark="checkLocalTransactionStateException:"+RemotingHelper.exceptionSimpleDesc(exception);}doExecuteEndTransactionHook(msg,uniqueKey,brokerAddr,localTransactionState,true);try{//再次口头endTransactionOneway动员END_TRANSACTIONDefaultMQProducerImpl.this.mQClientFactory.getMQClientAPIImpl().endTransactionOneway(brokerAddr,thisHeader,remark,3000);}catch(Exceptione){log.error("endTransactionOnewayexception",e);}}};this.checkExecutor.submit(request);}

总结

首先客户端Producer经过sendMessageInTransaction方法发送事务信息，Broker判别是事务信息就将信息topic存入到RMQ_SYS_TRANS_HALF_TOPIC前往给客户端，客户端继续口头逻辑。

而后调用endTransaction方法去提交本地事务经过endTransactionOneway将信息提交给Broker端，Broker端经过Code为END_TRANSACTION的处置器去处置信息调用processRequest方法来处置对应的信息，

假设由于各种要素造成信息的失败传输，为了防止这些现象的出现所以在BrokerController启动时就启动一个线程每隔6s处置未回查的信息（审核最大次数为15次）的义务来启动信息的回查，繁难来说就是经过sendCheckMessage方法去注册一个Code为CHECK_TRANSACTION_STATE的信息将内容发送给客户端，而后客户端在启动时也注册对应Code的处置逻辑，经过processTransactionState方法去处置事务的形态，假设反常最后还是会去口头endTransactionOneway方法，成功事务信息。

微服务架构的分布式事务问题如何处理？

分布式系统架构中，分布式事务问题是一个绕不过去的挑战。而微服务架构的流行，让分布式事问题日益突出！

下面我们以电商购物支付流程中，在各大参与者系统中可能会遇到分布式事务问题的场景进行详细的分析！

如上图所示，假设三大参与平台（电商平台、支付平台、银行）的系统都做了分布式系统架构拆分，按上数中的流程步骤进行分析：

1、电商平台中创建订单：预留库存、预扣减积分、锁定优惠券，此时电商平台内各服务间会有分布式事务问题，因为此时已经要跨多个内部服务修改数据；

2、支付平台中创建支付订单（选银行卡支付）：查询账户、查询限制规则，符合条件的就创建支付订单并跳转银行，此时不会有分布式事务问题，因为还不会跨服务改数据；

3、银行平台中创建交易订单：查找账户、创建交易记录、判断账户余额并扣款、增加积分、通知支付平台，此时也会有分布式事务问题（如果是服务化架构的话）；

4、支付平台收到银行扣款结果：更改订单状态、给账户加款、给积分帐户增加积分、生成会计分录、通知电商平台等，此时也会有分布式事务问题；

5、电商平台收到支付平台的支付结果：更改订单状态、扣减库存、扣减积分、使用优惠券、增加消费积分等，系统内部各服务间调用也会遇到分布式事问题；

如上图，支付平台收到银行扣款结果后的内部处理流程：

1、支付平台的支付网关对银行通知结果进行校验，然后调用支付订单服务执行支付订单处理；

2、支付订单服务根据银行扣款结果更改支付订单状态；

3、调用资金账户服务给电商平台的商户账户加款（实际过程中可能还会有各种的成本计费；如果是余额支付，还可能是同时从用户账户扣款，给商户账户加款）；

4、调用积分服务给用户积分账户增加积分；

5、调用会计服务向会计（财务）系统写进交易原始凭证生成会计分录；

6、调用通知服务将支付处理结果通知电商平台；

如上图，把支付系统中的银行扣款成功回调处理流程提取出来，对应的分布式事务问题的代码场景：

/** 支付订单处理 **/

@Transactional(rollbackFor = )

public void completeOrder() {

(); // 订单服务本地更新订单状态

(); // 调用资金账户服务给资金帐户加款

(); // 调用积分服务给积分帐户增加积分

(); // 调用会计服务向会计系统写入会计原始凭证

(); // 调用商户通知服务向商户发送支付结果通知

本地事务控制还可行吗？

以上分布式事务问题，需要多种分布式事务解决方案来进行处理。

订单处理：本地事务

资金账户加款、积分账户增加积分：TCC型事务（或两阶段提交型事务），实时性要求比较高，数据必须可靠。

会计记账：异步确保型事务（基于可靠消息的最终一致性，可以异步，但数据绝对不能丢，而且一定要记账成功）

商户通知：最大努力通知型事务（按规律进行通知，不保证数据一定能通知成功，但会提供可查询操作接口进行核对）

seata分布式事务原理是什么？

Seata框架是一个业务层的XA（两阶段提交）解决方案。在理解Seata分布式事务机制前，我们先回顾一下数据库层面的XA方案。

Seata 是一款开源的分布式事务解决方案，致力于提供高性能和简单易用的分布式事务服务。Seata 将为用户提供了 AT、TCC、SAGA 和 XA 事务模式，为用户打造一站式的分布式解决方案。

Transaction Coordinator (TC)： 事务协调器，维护全局事务的运行状态，负责协调并驱动全局事务的提交或回滚。

Transaction Manager (TM)： 控制全局事务的边界，负责开启一个全局事务，并最终发起全局提交或全局回滚的决议。