Spring Boot 事务实战之如何解决 DB 与 MQ 的"双写不一致"问题

// ❌ 错误示范
natsPublisher.publish(task); // 1. 消息发出，Worker 开始下载转存
messageRepository.save(message); // 2. 数据库报错（如字段超长、唯一键冲突）
// 后果：DB 回滚，业务无记录，但 OSS 中产生了一个永远无法被引用的“孤儿文件”。

// ❌ 错误示范
@Transactional
public void handle() {
    messageRepository.save(message);
    natsPublisher.publish(task); 
    // 3. 代码执行完毕，但在事务提交（Commit）的一瞬间数据库连接断开
}
// 后果：Worker 收到任务并完成处理，但在回调更新状态时发现 DB 中不存在该记录。

// 1. 准备阶段：预生成任务（纯内存操作，无副作用）
// 此时并没有真正发送 NATS 消息，只是构建了对象
List<MediaTransferTask> mediaTasks = prepareMediaTransferTasks(msg, ids.sessionId());
// 2. 构建消息实体
WxMessage message = buildMessage(msg, accountId, ids.sessionId(), ids.senderId());
try {
    // 【核心步骤 A】数据库落库 (Source of Truth)
    // 这是唯一的“事实来源”。如果这里失败，后续一切都不应发生。
    messageRepository.save(message);
    log.info("Message saved: id={}, wxid={}", message.getId(), message.getWxid());
    // 3. 发布会话更新事件 (内存事件或 MQ)
    SessionUpdateEvent event = SessionUpdateEvent.builder()
        .accountId(accountId)
        // ... build params
        .build();
    sessionEventPublisher.publishSessionUpdate(event);
    // 【核心步骤 B】注册事务回调
    // 关键点：这里不是立即发送，而是“预约”发送
    publishMediaTransferTasksAfterCommit(mediaTasks);
    return ProcessResult.success();
} catch (DataIntegrityViolationException e) {
    // 【并发场景的保护】
    // 如果两个线程同时处理同一条消息（如网络重放或客户端重试），
    // 数据库的唯一索引会抛出此异常。
    // 由于消息发送逻辑在事务提交后执行，失败的线程事务回滚，
    // 因此“afterCommit”钩子不会被触发，完美避免了 Worker 重复搬运文件。
    log.debug("Duplicate message (concurrent): {}", msg.getMessageId());
    return ProcessResult.duplicate();
}

publishMediaTransferTasksAfterCommit 方法利用了 Spring 的 TransactionSynchronizationManager 来挂载回调。
private void publishMediaTransferTasksAfterCommit(List<MediaTransferTask> tasks) {
    if (CollectionUtils.isEmpty(tasks)) {
        return;
    }
    // 判断当前是否在事务中
    if (TransactionSynchronizationManager.isActualTransactionActive()) {
        // 注册同步器
        TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
            @Override
            public void afterCommit() {
                // 【真正发送的时机】
                // 只有当 DB 事务成功 Commit 后，这一行才会执行
                // 此时 DB 里一定有数据，Worker 回调一定能成功
                tasks.forEach(mediaTransferPublisher::publishMediaTransfer);
                log.debug("Async media tasks published after commit: size={}", tasks.size());
            }
        });
    } else {
        // 如果不在事务中（比如非事务方法调用），则立即发送（降级策略）
        tasks.forEach(mediaTransferPublisher::publishMediaTransfer);
    }
}