在当今的微服务架构中，分布式事务管理是一个核心问题。RocketMQ，作为阿里巴巴开源的一款分布式消息中间件，为解决这一难题提供了强有力的支持。本篇文章将详细介绍如何在实战中使用 RocketMQ 实现分布式事务消息，同时包含相应的代码示例。
一、RocketMQ 分布式事务简介
RocketMQ 支持消息的完全顺序性、消息的幂等性以及高可用性。在分布式系统中，由于网络分区、节点故障等原因，有时会出现部分服务提交、部分服务失败的情况，导致数据不一致。RocketMQ 的分布式事务消息功能可以在这种情况下保证数据的完整性。
二、RocketMQ 分布式事务消息实现
- 配置 RocketMQ Broker
首先，需要在 RocketMQ Broker 的配置文件中开启事务消息功能。具体配置如下：
# 开启事务消息功能 transaction.message.Enable=true
- 生产者发送事务消息
在生产者端，我们需要使用 RocketMQ 的 TransactionMQProducer 来发送事务消息。以下是一个简单的示例代码：
public class Producer { public static void main(String[] args) throws Exception { // 创建生产者实例 DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName"); producer.setNamesrvAddr("127.0.0.1:9876"); // 设置 RocketMQ Broker 地址 producer.start(); for (int i = 0; i < 10; i++) { // 创建事务消息请求 TransactionMQProducer producer1 = new TransactionMQProducer("ProducerGroupName"); producer1.setNamesrvAddr("127.0.0.1:9876"); // 设置 RocketMQ Broker 地址 producer1.setRetryTimesWhenSendAsyncFailed(3); // 设置重试次数 producer1.setMessageValidator(new MessageValidatorImpl()); // 设置自定义的消息验证器 producer1.start(); TransactionSendResult result = producer1.send(buildTransactionMessage("TopicTest", "TagA", "OrderID" + i), new MessageQueueSelector() { @Override public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) { // 根据业务规则选择一个 MessageQueue 进行发送 return mqs.get(0); } }); // 等待消息发送结果，如果发送失败则回滚操作，否则提交操作。 if (result.getSendStatus() == SendStatus.SEND_ERROR) { System.out.println("Send Error, roll back action"); // 进行回滚操作... producer1.retryMessageQueue(result.getMsgQueue(), result.getMsg()); } else { System.out.println("Send Success, commit action"); // 进行提交操作... producer1.commitMessageQueue(result.getMsgQueue(), result.getMsg()); } producer1.shutdown(); // 关闭生产者实例 } producer.shutdown(); // 关闭生产者实例 } }
- 消费者处理事务消息
在消费者端，我们需要使用 RocketMQ 的 TransactionalConsumer 来消费事务消息。以下是一个简单的示例代码：
public class Consumer { public static void main(String[] args) throws Exception { // 创建消费者实例并订阅主题和标签 DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ConsumerGroupName"); consumer.subscribe("TopicTest", "*"); // 订阅主题和标签 consumer.setNamesrvAddr("127.0.0.1:9876"); // 设置 RocketMQ Broker 地址 consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> { for (MessageExt msg : msgs) { // 处理消息列表 System.out.println("Receive message: " + msg); // 打印收到的消息内容 try { // 在业务代码块中完成业务逻辑处理，并在最后调用 confirm 方法提交或 reject 方法回滚。若业务处理失败或需要回滚，则抛出异常。当业务代码执行成功，但无法调用 confirm 或 reject 时，可以在 catch 块中抛出业务异常。这样 RocketMQ 会自动将该消息放到死信队列中，供其他消费者处理。死信队列中的消息优先级高于普通队列。默认情况下，死信队列名为原队列名后加一个点，例如，原队列名为“test_queue”，则死信队列名为“test_queue.DLQ”。 ```java try { // 业务逻辑处理... System.out.println("Message processing success, commit it."); // 提交消息 msg.setReconsumeTimes(0); // 重试次数清零 context.getConsumer().confirm(msg); } catch (Exception e) { System.out.println("Message processing error, roll back it."); // 回滚消息 msg.setReconsumeTimes(5); // 设置重试次数为5 context.getConsumer().reject(msg); } }); consumer.start(); // 启动消费者实例 } }
三、注意事项
- 在生产者发送事务消息时，需要保证网络连接的稳定性，避免出现网络分区、延迟等问题。
- 在消费者处理事务消息时，需要保证业务逻辑的正确性和健壮性，避免出现异常导致消息无法提交或回滚。
- 在生产者发送事务消息时，可以根据业务需要自定义消息验证器，对消息内容进行校验，确保消息的正确性和合法性。
- 在消费者处理事务消息时，可以根据业务需要设置不同的重试次数和回滚策略，以实现不同场景下的容错处理。
- 在使用 RocketMQ 分布式事务消息时，需要注意消息的顺序性和幂等性，避免出现重复处理或遗漏处理的情况。
