Spring 事务与 @Async 导致的状态丢失

1. 背景

最近在做任务发布功能的时候，线上遇到一个 玄学 Bug。

场景

用户点击”发布任务”后，后台日志显示发布成功，但任务状态偶尔还停留在”发布中”，没有变成”已发布”。

经过初步排查发现：

• 任务发布是一个 异步流程
• 用户点击发布后，主线程先把状态改为 发布中（PUBLISHING）
• 然后通过 @Async 触发异步线程去做真正的发布（渲染 DAG、上传 Airflow、轮询确认等）
• 异步线程执行完毕后，把状态改为 已发布（PUBLISHED）

从日志来看：

诡异现象

异步线程明明打了”异步发布任务成功”的日志，数据库里的状态却还是 PUBLISHING 🤯

2. 初始实现思路

先看一下发布流程的核心代码结构。

主线程（TaskServiceImpl.publishTask）：

@Override
@Transactional
public boolean publishTask(Long taskId) {
    // ... 各种校验 ...

    // 状态改为【发布中】
    TaskPO update = new TaskPO();
    update.setId(taskId);
    update.setStatus(TaskStatusEnum.PUBLISHING);
    update.setPublishBy(UserContextHolder.getUserId());
    update.setPublishTime(LocalDateTime.now());
    taskMapper.updateById(update);

    // 触发异步发布
    asyncPublishTask(taskId, db.getTaskType());

    return true;
}

异步线程（AsyncTaskPublishExecutor.publish）：

@Async("taskPublishExecutor")
public void publish(Long taskId) {
    TaskPO task = taskMapper.selectById(taskId);

    // 执行发布（渲染 DAG → SFTP 上传 → 轮询 Airflow → 确认上线）
    handlerFactory.getHandler(task.getTaskType()).execute(task);

    // 成功 → 已发布
    TaskPO update = new TaskPO();
    update.setId(taskId);
    update.setStatus(TaskStatusEnum.PUBLISHED);
    taskMapper.updateById(update);
}

逻辑很清晰，对吧？先 PUBLISHING，再异步执行，成功了改 PUBLISHED。

本地调试、Postman 单次请求 一切正常。

然而——

一上环境就偶发翻车了。

3. 线上问题现象

在测试环境反复测试后发现：

• 大部分时候发布正常
• 偶尔 会出现状态卡在 PUBLISHING
• 日志里异步线程确实打了”异步发布任务成功”
• 但数据库查出来就是 PUBLISHING

进一步对比日志时间戳后发现一个关键线索：

关键现象

异步线程的 updateById(PUBLISHED) 和主线程的事务提交，时间非常接近，有时候甚至是毫秒级的先后关系。

4. 关键问题定位

冷静下来画了一下时序图，问题一下就清晰了：

致命问题

publishTask() 方法上标了 @Transactional，事务的提交时机是方法返回之后，由 Spring AOP 代理完成。而 @Async 在方法体内被调用时，异步线程已经提交到线程池开始执行了，但主线程的事务还没提交。

这就导致了一个经典的竞态：

• 正常情况：Airflow 部署流程比较耗时（SFTP 上传 + 轮询等待最多 45 秒），主线程事务早就提交了，异步线程后写的 PUBLISHED 不会被覆盖。所以大部分时候是正常的。
• 翻车情况：如果 Airflow 响应特别快（DAG 已经存在、扫描秒过），异步线程可能在主线程事务提交 之前 就完成了 UPDATE status = PUBLISHED。然后主线程事务提交时，把 status = PUBLISHING 又刷回去了。

这也解释了为什么：

• “大部分时候正常” —— 因为 Airflow 通常不会秒回
• “偶尔翻车” —— 恰好 Airflow 响应很快的时候

核心结论

这不是 Airflow 轮询的问题，而是 Spring @Transactional 与 @Async 配合使用时，事务提交时机与异步线程执行时机之间的竞态条件。

5. 为什么会被覆盖？

可能有人会问：异步线程先写了 PUBLISHED，主线程后写 PUBLISHING，数据库不会有锁保护吗？

这里要注意两点：

5.1 MyBatis-Plus 的 updateById 行为

taskMapper.updateById(update) 生成的 SQL 大致是：

UPDATE t_kaas_task SET fstatus = 'PUBLISHING', fupdated_by = ?, fupdated_time = ? WHERE fid = ? AND fdeleted = false

它是一个 无条件的 SET，不会检查当前状态是什么，直接覆盖。

5.2 没有乐观锁

TaskPO 上没有 @Version 注解，也就是说 MyBatis-Plus 不会在 WHERE 条件里加版本号校验。谁最后提交，谁的值就生效。

5.3 事务隔离级别

MySQL 默认的 REPEATABLE READ 隔离级别下，主线程事务内的 UPDATE 在事务提交时才真正持久化。如果异步线程的 UPDATE 先落盘了，主线程事务提交时会直接用自己的值覆盖掉。

三者叠加，后提交的事务 无声无息地 把正确的状态覆盖了。

6. 最终解决方案

最终方案

使用 Spring 的 TransactionSynchronizationManager 注册 afterCommit 回调，确保异步发布在主线程事务提交之后才触发。

6.1 核心目标

• 保证 status = PUBLISHING 已经持久化到数据库 之后，才启动异步发布
• 异步线程写入的 PUBLISHED 不会被任何后续操作覆盖

6.2 修复代码

@Override
@Transactional
public boolean publishTask(Long taskId) {
    try {
        // ... 校验逻辑不变 ...

        // 状态改为【发布中】
        TaskPO update = new TaskPO();
        update.setId(taskId);
        update.setStatus(TaskStatusEnum.PUBLISHING);
        update.setPublishBy(UserContextHolder.getUserId());
        update.setPublishTime(LocalDateTime.now());
        taskMapper.updateById(update);

        // 关键修复：事务提交后再触发异步发布
        TaskTypeEnum taskType = db.getTaskType();
        TransactionSynchronizationManager.registerSynchronization(
            new TransactionSynchronization() {
                @Override
                public void afterCommit() {
                    log.info("事务已提交, 开始异步发布任务, taskId={}", taskId);
                    asyncPublishTask(taskId, taskType);
                }
            }
        );

    } catch (BaseException e) {
        log.error("任务发布异常! 任务ID: {}, 错误详情: {}", taskId, e.getMessage(), e);
        taskMapper.updateById(TaskPO.ofPublishFailed(taskId, e));
    } catch (Exception e) {
        log.error("任务发布异常! 任务ID: {}, 错误详情: {}", taskId, e.getMessage(), e);
        BaseException unknownEx = new BaseException(
                TaskPublishErrorCode.UNKNOWN_ERROR.getCode(),
                "系统内部错误: " + e.getMessage(), e);
        taskMapper.updateById(TaskPO.ofPublishFailed(taskId, unknownEx));
    }
    return true;
}

6.3 修复后的时序

现在事务提交和异步执行之间有了明确的 先后顺序保证，不再依赖”Airflow 够不够慢”这种玄学条件。

7. 总结

这个 Bug 表面上看是：

表象问题

任务发布成功但状态没更新

但本质上是：

• @Transactional 的事务提交时机是 方法返回后
• @Async 的异步线程是 方法体内就启动了
• 两者之间没有先后顺序保证
• MyBatis-Plus updateById 无条件覆盖 + 无乐观锁 = 后提交者赢

最终的经验总结：

1. @Transactional 方法内不要直接调用 @Async 方法，除非你能保证异步操作不依赖当前事务的提交结果
2. 需要”事务提交后再做某事”的场景，用 TransactionSynchronizationManager.registerSynchronization 的 afterCommit 回调
3. 这类 Bug 的恶心之处在于 大部分时候正常，只在特定时序下才复现，本地调试几乎不可能触发
4. 涉及状态流转的 UPDATE，考虑加 乐观锁（@Version） 或 WHERE 条件校验当前状态，作为最后一道防线

吐槽

Spring 也不给个编译期警告，@Transactional 里调 @Async 这种经典坑，每年不知道要坑多少人 🙃