RocketMQ事务消息源码分析

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消息,如果有说明该条半消息状态已确定,不需要回查,否则在判断是否已经到了最大回查次数和回查过期时间,如果这些都满足,则进行回查:

  1. 将半消息复制一份再放回队列中,以便后续回查
  2. 执行回查逻辑,根据生产组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
    }
}
最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 215,723评论 6 498
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 92,003评论 3 391
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 161,512评论 0 351
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 57,825评论 1 290
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 66,874评论 6 388
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 50,841评论 1 295
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 39,812评论 3 416
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 38,582评论 0 271
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 45,033评论 1 308
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 37,309评论 2 331
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 39,450评论 1 345
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 35,158评论 5 341
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 40,789评论 3 325
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 31,409评论 0 21
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 32,609评论 1 268
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 47,440评论 2 368
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 44,357评论 2 352

推荐阅读更多精彩内容