如何使用rocketmq实现分布式事务？

什么是rocketmq事务消息

事务消息是 Apache RocketMQ 提供的一种高级消息类型，支持在分布式场景下保障消息生产和本地事务的最终一致性。

RocketMQ的分布式事务又称为“半消息事务”。

事务消息处理流程

RocketMQ是靠半消息机制实现分布式事务

事务消息：MQ 提供类似 X/Open XA 的分布事务功能，通过 MQ 事务消息能达到分布式事务的最终一致。

半消息：暂不能投递的消息，发送方已经将消息成功发送到了 MQ 服务端，但是服务端未收到生产者对该消息的二次确认，此时该消息被标记成“暂不能投递”状态，处于该种状态下的消息即半消息。

半消息回查：由于网络闪断、生产者应用重启等原因，导致某条事务消息的二次确认丢失，MQ 服务端通过扫描发现某条消息长期处于“半消息”时，需要主动向消息生产者询问该消息的最终状态（Commit 或是 Rollback），该过程即消息回查。

事务消息交互流程如下图所示。

1. 生产者将消息发送至Apache RocketMQ服务端。

2. Apache RocketMQ服务端将消息持久化成功之后，向生产者返回Ack确认消息已经发送成功，此时消息被标记为"暂不能投递"，这种状态下的消息即为半事务消息。

3. 生产者开始执行本地事务逻辑。

4. 生产者根据本地事务执行结果向服务端提交二次确认结果（Commit或是Rollback），服务端收到确认结果后处理逻辑如下：

• 二次确认结果为Commit：服务端将半事务消息标记为可投递，并投递给消费者。

• 二次确认结果为Rollback：服务端将回滚事务，不会将半事务消息投递给消费者。

5. 在断网或者是生产者应用重启的特殊情况下，若服务端未收到发送者提交的二次确认结果，或服务端收到的二次确认结果为Unknown未知状态，经过固定时间后，服务端将对消息生产者即生产者集群中任一生产者实例发起消息回查。说明服务端回查的间隔时间和最大回查次数，请参见参数限制。

6. 生产者收到消息回查后，需要检查对应消息的本地事务执行的最终结果。

7. 生产者根据检查到的本地事务的最终状态再次提交二次确认，服务端仍按照步骤4对半事务消息进行处理。

事务消息生命周期

事务消息

• 初始化：半事务消息被生产者构建并完成初始化，待发送到服务端的状态。

• 事务待提交：半事务消息被发送到服务端，和普通消息不同，并不会直接被服务端持久化，而是会被单独存储到事务存储系统中，等待第二阶段本地事务返回执行结果后再提交。此时消息对下游消费者不可见。

• 消息回滚：第二阶段如果事务执行结果明确为回滚，服务端会将半事务消息回滚，该事务消息流程终止。

• 提交待消费：第二阶段如果事务执行结果明确为提交，服务端会将半事务消息重新存储到普通存储系统中，此时消息对下游消费者可见，等待被消费者获取并消费。

• 消费中：消息被消费者获取，并按照消费者本地的业务逻辑进行处理的过程。此时服务端会等待消费者完成消费并提交消费结果，如果一定时间后没有收到消费者的响应，Apache RocketMQ会对消息进行重试处理。

• 消费提交：消费者完成消费处理，并向服务端提交消费结果，服务端标记当前消息已经被处理（包括消费成功和失败）。Apache RocketMQ默认支持保留所有消息，此时消息数据并不会立即被删除，只是逻辑标记已消费。消息在保存时间到期或存储空间不足被删除前，消费者仍然可以回溯消息重新消费。

• 消息删除：Apache RocketMQ按照消息保存机制滚动清理最早的消息数据，将消息从物理文件中删除。

示例

下面是使用 RocketMQ 实现事务的一个例子：

生产者实现事务监听器：

首先，需要实现一个 RocketMQ 的事务监听器接口RocketMQLocalTransactionListener，这个接口定义了在发送和确认事务消息时的回调方法。您需要根据业务逻辑来实现这些方法。

executeLocalTransaction 方法：

这个方法在发送事务消息时被调用，用于执行本地事务。具体步骤如下：

1. 获取消息中的事务 ID。

2. 根据事务索引来模拟本地事务执行的状态。

3. 将事务状态放入 localTrans 映射中，以备后续 checkLocalTransaction 方法使用。

根据您的代码，executeLocalTransaction 方法中模拟了三种状态：

• 如果状态为 0，表示本地事务成功，返回 RocketMQLocalTransactionState.COMMIT，消息将被提交。

• 如果状态为 1，表示本地事务失败，返回 RocketMQLocalTransactionState.ROLLBACK，消息将被回滚。

• 如果状态为 2，表示本地事务状态未知，返回 RocketMQLocalTransactionState.UNKNOWN。

checkLocalTransaction 方法：

这个方法在消息的确认状态时被调用，用于检查本地事务的状态。具体步骤如下：

获取消息中的事务 ID。
根据之前保存在 localTrans 映射中的事务状态，决定消息的提交、回滚或未知。

checkLocalTransaction 方法会根据之前在 executeLocalTransaction 方法中保存的状态来返回相应的事务状态。

@RocketMQTransactionListener  
public class TransactionListenerImpl implements RocketMQLocalTransactionListener {  private AtomicInteger transactionIndex = new AtomicInteger(0);  private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<String, Integer>();  @Override  public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {  String transId = (String) msg.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);  System.out.printf("#### executeLocalTransaction is executed, msgTransactionId=%s %n",  transId);  int value = transactionIndex.getAndIncrement();  int status = value % 3;  localTrans.put(transId, status);  if (status == 0) {  // Return local transaction with success(commit), in this case,  // this message will not be checked in checkLocalTransaction()  System.out.printf(" # COMMIT # Simulating msg %s related local transaction exec succeeded! ### %n", msg.getPayload());  return RocketMQLocalTransactionState.COMMIT;  }  if (status == 1) {  // Return local transaction with failure(rollback) , in this case,  // this message will not be checked in checkLocalTransaction()  System.out.printf(" # ROLLBACK # Simulating %s related local transaction exec failed! %n", msg.getPayload());  return RocketMQLocalTransactionState.ROLLBACK;  }  System.out.printf(" # UNKNOW # Simulating %s related local transaction exec UNKNOWN! \n");  return RocketMQLocalTransactionState.UNKNOWN;  }  @Override  public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {  String transId = (String) msg.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);  RocketMQLocalTransactionState retState = RocketMQLocalTransactionState.COMMIT;  Integer status = localTrans.get(transId);  if (null != status) {  switch (status) {  case 0:  retState = RocketMQLocalTransactionState.COMMIT;  break;  case 1:  retState = RocketMQLocalTransactionState.ROLLBACK;  break;  case 2:  retState = RocketMQLocalTransactionState.UNKNOWN;  break;  }  }  System.out.printf("------ !!! checkLocalTransaction is executed once," +  " msgTransactionId=%s, TransactionState=%s status=%s %n",  transId, retState, status);  return retState;  }  
}

消费者

@Service  
@RocketMQMessageListener(topic = "${demo.rocketmq.transTopic}", consumerGroup = "string_trans_consumer")  
public class StringTransactionalConsumer implements RocketMQListener<String> {  @Override  public void onMessage(String message) {  System.out.printf("------- StringTransactionalConsumer received: %s \n", message);  }  
}

这些步骤基本上涵盖了使用 RocketMQ 实现事务的主要过程。可以根据具体的业务需求和环境进行调整和配置。