RocketMQ 事务消息（Transactional Message）是指
应用本地事务和发送消息操作可以被定义到全局事务中，要么同时成功，要么同时失败。
只要发送OK了，RocketMQ 会保证消息被 至少消费一次，从而保证上下游数据一致。
RocketMQ 采用了 2PC 的思想来实现了提交事务消息，同时增加一个补偿逻辑来处理二阶段超时或者失败的消息。具体如下如所示：
image.png
执行原理
- producer 提交半消息到 broker 半消息 topic，真实消息内容存储在半消息的扩展属性中；
- 如果半消息发送成功，则执行本地事务，如果半消息没发送成功，则不执行本地事务，此时，半消息的发送可以使用重试机制来解决；
- 当本地事务执行成功后，producer 根据本地事务的执行状态发送 commit 或者 rollback 消息给 broker，broker 接收到 rollback 消息后，直接抛弃半消息；接收到 commit 消息后，从半消息的扩展属性中解析出真正的消息，推到真实的 topic 和 queue 中（消息存储成功后，删除半消息），后续 consumer 进行消费；
- 如果 producer 发送 commit 或 rollback 消息失败了，broker 会针对这些未确定状态的半消息调用 producer 进行回查，producer 查询本地事务执行情况，之后根据本地事务的执行状态发送 commit 或者 rollback 消息给 broker。
值得注意的是，RocketMQ 并不会无休止的的信息事务状态回查，默认回查15次，如果15次回查还是无法得知事务状态，RocketMQ 默认回滚该消息。
事务监听器
/**
* 事务监听器（执行本地事务/回查本地事务执行状态）
*/
public class TransactionListenerImpl implements org.apache.rocketmq.client.producer.TransactionListener {
private AtomicInteger transactionIndex = new AtomicInteger(0);
/**
* 记录本地事务ID<->本地事务处理状态
* 用于补偿链路的回查操作
*/
private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();
/**
* 执行本地事务逻辑
*
* @param msg Half(prepare) message
* @param arg Custom business parameter
* @return
*/
@Override
public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
int value = transactionIndex.getAndIncrement();
int status = value % 3;
localTrans.put(msg.getTransactionId(), status);
/**
* 返回事务状态，
* COMMIT_MESSAGE - 本地事务处理成功
* ROLLBACK_MESSAGE - 本地事务处理失败
* UNKNOW - 在这里仅用于 mock，这样的返回，会触发回查逻辑
*/
return LocalTransactionState.UNKNOW;
}
/**
* 回查本地事务执行状态
*
* @param msg Check message
* @return
*/
@Override
public LocalTransactionState checkLocalTransaction(MessageExt msg) {
System.out.println("回查");
Integer status = localTrans.get(msg.getTransactionId());
if (null != status) {
switch (status) {
case 0:
return LocalTransactionState.UNKNOW;
case 1:
return LocalTransactionState.COMMIT_MESSAGE;
case 2:
return LocalTransactionState.ROLLBACK_MESSAGE;
default:
return LocalTransactionState.COMMIT_MESSAGE;
}
}
return LocalTransactionState.COMMIT_MESSAGE;
}
}
生产者
public class TransactionProducer {
public static void main(String[] args) throws MQClientException, InterruptedException {
TransactionListener transactionListener = new TransactionListenerImpl();
TransactionMQProducer producer = new TransactionMQProducer("please_rename_unique_group_name");
producer.setNamesrvAddr("127.0.0.1:9876");
// https://blog.csdn.net/heihaozi/article/details/119145266
DefaultChannelId.newInstance();
// 设置事务监听器（执行本地事务/回查本地事务执行状态）
producer.setTransactionListener(transactionListener);
producer.start();
String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};
for (int i = 0; i < 10; i++) {
try {
Message msg = new Message("TopicTest", tags[i % tags.length], ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
// 这里的 sendMessageInTransaction 实际上做了上图中的1,2,3,4。
TransactionSendResult sendResult = producer.sendMessageInTransaction(msg, null);
// 消息发送成功，并且本地事务执行状态不是回滚
if (SendStatus.SEND_OK.equals(sendResult.getSendStatus()) && !LocalTransactionState.ROLLBACK_MESSAGE.equals(sendResult.getLocalTransactionState())) {
System.out.printf("%s%n", sendResult);
}
Thread.sleep(10);
} catch (MQClientException | UnsupportedEncodingException e) {
e.printStackTrace();
}
}
for (int i = 0; i < 100000; i++) {
Thread.sleep(1000);
}
producer.shutdown();
}
}
一个完整的案例见：https://github.com/dingwpmz/rocketmq-learning
