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
}
}