分布式事务是一个复杂的问题，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)
事务消息的发送过程和普通消息发送过程是不一样的，发送消息的方法是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; }