rocketMQ从4.3.0版本开始支持事务消息，事务消息能确保生产者本地事务和消息发送的的原子性。

具体流程

半消息落盘

在执行本地事务前，先同步发送一条半消息，存储在系统主题RMQ_SYS_TRANS_HALF_TOPIC中，对消费者不可见

事务消息不支持批量发送和延时消息

// ignore DelayTimeLevel parameter
if (msg.getDelayTimeLevel() != 0) {
    MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_DELAY_TIME_LEVEL);
}

半消息的PROPERTY_TRANSACTION_PREPARED属性是true，并且会设置生产者组，作用后面讲

MessageAccessor.putProperty(msg, MessageConst.PROPERTY_TRANSACTION_PREPARED, "true");
MessageAccessor.putProperty(msg, MessageConst.PROPERTY_PRODUCER_GROUP, this.defaultMQProducer.getProducerGroup());

broker端会判断如果PROPERTY_TRANSACTION_PREPARED为true，则将它的真实主题和队列ID退避到属性里面，并把主题设置为RMQ_SYS_TRANS_HALF_TOPIC，队列ID设置为0

String transFlag = origProps.get(MessageConst.PROPERTY_TRANSACTION_PREPARED);
if (transFlag != null && Boolean.parseBoolean(transFlag)) {
    ...
}

MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_REAL_TOPIC, msgInner.getTopic());
MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_REAL_QUEUE_ID,
                            String.valueOf(msgInner.getQueueId()));
//半消息sysflag为TRANSACTION_NOT_TYPE
msgInner.setSysFlag(
    MessageSysFlag.resetTransactionValue(msgInner.getSysFlag(), MessageSysFlag.TRANSACTION_NOT_TYPE));
msgInner.setTopic(TransactionalMessageUtil.buildHalfTopic());
msgInner.setQueueId(0);

END_TRANSACTION请求

本地事务执行完后，发送END_TRANSACTION请求提交3种状态，提交、回滚或未知，提交和回滚都会查找半消息，并且将它逻辑删除，提交还会根据原半消息生成一条新的消息，还原真实主题和队列ID，落盘。未知状态则不做任何处理，以便后续回查

//提交，还原真实主题和队列ID
msgInner.setTopic(msgExt.getUserProperty(MessageConst.PROPERTY_REAL_TOPIC));
msgInner.setQueueId(Integer.parseInt(msgExt.getUserProperty(MessageConst.PROPERTY_REAL_QUEUE_ID)));
//清空属性
MessageAccessor.clearProperty(msgInner, MessageConst.PROPERTY_REAL_TOPIC);
MessageAccessor.clearProperty(msgInner, MessageConst.PROPERTY_REAL_QUEUE_ID);
MessageAccessor.clearProperty(msgInner, MessageConst.PROPERTY_TRANSACTION_PREPARED);

逻辑删除

本质上是生成一条OP消息，主题是RMQ_SYS_TRANS_OP_HALF_TOPIC，tags是"d"，消息体是对应删除的半消息的逻辑偏移量（queue offset），以便后续回查

Message message = new Message(TransactionalMessageUtil.buildOpTopic(), TransactionalMessageUtil.REMOVETAG,
                              String.valueOf(messageExt.getQueueOffset()).getBytes(TransactionalMessageUtil.charset));

回查

回查有3个可配置的属性：

transactionCheckInterval：消息回查定时任务的时间间隔，默认60s
transactionTimeOut：事务过期时间，表示第一次回查的最短时间间隔，从生产时间开始计时，可以被消息属性CHECK_IMMUNITY_TIME_IN_SECONDS覆盖，默认6s
transactionCheckMax：最大回查次数，消息的回查次数存储在消息属性TRANSACTION_CHECK_TIMES中，每次回查会自增，默认15次，超过会存到TRANS_CHECK_MAX_TIME_TOPIC主题中

消息回查是通过broker端的一个定时任务TransactionalMessageCheckService去消费半消息，判断半消息是否有对应的op消息，如果有说明该条半消息状态已确定，不需要回查，否则在判断是否已经到了最大回查次数和回查过期时间，如果这些都满足，则进行回查：

将半消息复制一份再放回队列中，以便后续回查
执行回查逻辑，根据生产组ID获取channel，broker作为客户端发送CHECK_TRANSACTION_STATE请求，客户端使用回查线程池执行回查逻辑，根据返回状态再发送END_TRANSACTION请求

ServiceThread

RMQ中间隔执行的任务类都是通过继承抽象类ServiceThread实现的，比如事务消息回查任务TransactionalMessageCheckService，长轮询PullRequestHoldService等

public class TransactionalMessageCheckService extends ServiceThread {
    @Override
    public void run() {
        long checkInterval = brokerController.getBrokerConfig().getTransactionCheckInterval();
        while (!this.isStopped()) {
            this.waitForRunning(checkInterval);
        }
    }

    @Override
    protected void onWaitEnd() {
//具体业务逻辑
    }

}

ServiceThread实现了Runnable，并持有了一个Thread，该Thread执行自身run方法

有如下几个特性：

start启动后，持有的线程会执行waitForRunning，先阻塞一段时间，再执行具体业务逻辑
wakeup唤醒等待在门闩的线程，这样既可以定时执行任务，又能通过显示调用方法提前执行任务，比如再平衡（rebalance)中消费者发生变化时，需要还没到超时时间就立刻触发一次再平衡，rebalance以后再详细讲
stop将终止状态位stopped置为true，用户可以判断该状态位来结束线程，stop也会提前唤醒等待CDL线程，支持中断线程
makestop只将stopped置为true，不唤醒
shutdown会把started置为false，stopped置为true，唤醒等待线程，支持中断线程，还会join等待线程执行完，超时时间默认90s

成员变量

private Thread thread：持有的线程，会使用该线程执行任务

protected final CountDownLatch2 waitPoint = new CountDownLatch2(1)：jdk提供的CDL是一次性的，await返回后，不能再使用，rmq在CountDownLatch的基础上增加了重置功能，初始化状态时，将初始状态值记录下来，reset时调用AQS的setState方法重新设置状态，从而可以重复使用

private static final class Sync extends AbstractQueuedSynchronizer {
    private final int startCount;
    Sync(int count) {
        this.startCount = count;
        setState(count);
    }
    protected void reset() {
        setState(startCount);
    }
    //省略其他代码
}

protected volatile AtomicBoolean hasNotified = new AtomicBoolean(false)：通知标志，和waitPoint 配合使用，每次countDown的同时开启通知标志，因为waitPoint 会重置，导致countDown无效，所以要在重置前判断通知是否开启，确保wakeup后会立即执行一次业务逻辑。如下代码，

如果1处wakeup，则正常唤醒；
如果2之前1之后wakeup，hasNotified会在2处重置，3不会多执行；
如果2之后,4之前wakeup，因为onWaitEnd可能已经执行了一部分，需要立即执行3
如果5之后执行wakeup，会导致较长时间等待，这里尚存疑问，只能解释为这里执行速度较快，发生概率很小

protected void waitForRunning(long interval) {
    if (hasNotified.compareAndSet(true, false)) {//4
        this.onWaitEnd();//3
        return;
    }//5

    //entry to wait
    waitPoint.reset();

    try {
        waitPoint.await(interval, TimeUnit.MILLISECONDS);//1
    } catch (InterruptedException e) {
        log.error("Interrupted", e);
    } finally {
        hasNotified.set(false);//2
        this.onWaitEnd();
    }
}

public void wakeup() {
    if (hasNotified.compareAndSet(false, true)) {
        waitPoint.countDown(); // notify
    }
}

RocketMQ事务消息源码分析