多线程批量插入如何保证事务一致性

问题背景

在实际业务中经常遇到这样的场景：有一批业务数据需要批量插入数据库，为了提高性能开了多个线程并发写入。但随之而来一个关键问题——如何保证事务的正确性，要么全部成功，要么全部失败？

为什么多线程事务这么难？

Spring 的 @Transactional 基于 ThreadLocal 实现，每个线程拥有独立的数据库连接和事务上下文，天然无法跨线程协调。

常见的”错误方案”：CountDownLatch 协调多线程提交

网上流传最广的方案是用两个 CountDownLatch 协调多线程事务：所有子线程执行完后等待主线程信号，主线程检查结果后统一通知提交或回滚。

这个方案有致命缺陷——commit 阶段不是原子的：

T1: conn1.commit()  →  成功 ✓  (已持久化，不可撤回)
T2: conn2.commit()  →  成功 ✓  (已持久化，不可撤回)
T3: conn3.commit()  →  网络断开 / 死锁 / 磁盘满  ✗

T1 和 T2 已经落盘，无法回滚。哪怕让所有子线程在同一时刻接到 commit 信号，也无法做到数据库层面的原子提交。commit 失败的原因（死锁、连接异常、存储错误、主库切换等）在提交瞬间才发生，无法预判。

核心矛盾：每个线程持有独立的数据库连接，每个 commit() 是一次独立的网络调用，多个独立 commit 的成功/失败不可能原子化。

---

方案总览

方案	原子性	性能	适用数据量	复杂度
方案一：单事务 BATCH 模式	强一致	高	1万内最佳，10万也可	低
方案二：多线程准备 + 单线程提交	强一致	更高（CPU密集场景）	同上	中
方案三：Staging 伪 2PC	强一致（业务表层面）	高	中等	中高
方案四：分批提交 + 补偿机制	最终一致	最高	百万级	高

---

方案一：单事务 + BATCH 模式（推荐）

核心思路

不用多线程提交，在 单个事务内 使用 MyBatis 的 ExecutorType.BATCH 模式优化插入性能。配合 MySQL 驱动的 rewriteBatchedStatements=true，单线程插入 1 万条数据通常在 1 秒以内。

为什么够快？

优化手段	效果
ExecutorType.BATCH	MyBatis 攒 SQL 批量发送，减少网络往返
rewriteBatchedStatements=true	MySQL 驱动合并多条 INSERT 为一条多 VALUES
单连接单事务	没有线程协调开销，没有连接争用

rewriteBatchedStatements=true 开启后，底层会把：

INSERT INTO t VALUES (1, 'a');
INSERT INTO t VALUES (2, 'b');
INSERT INTO t VALUES (3, 'c');

合并成：

INSERT INTO t VALUES (1, 'a'), (2, 'b'), (3, 'c');

示例代码

@Service
@RequiredArgsConstructor
public class SingleTxBatchService {

    private final SqlSessionFactory sqlSessionFactory;

    @Transactional(rollbackFor = Exception.class)
    public void batchInsert(List<OrderItem> dataList) {
        try (SqlSession session = sqlSessionFactory.openSession(ExecutorType.BATCH)) {
            OrderItemMapper mapper = session.getMapper(OrderItemMapper.class);

            for (int i = 0; i < dataList.size(); i++) {
                mapper.insert(dataList.get(i));
                if ((i + 1) % 1000 == 0) {
                    session.flushStatements();
                }
            }
            session.flushStatements();
        }
    }
}

优缺点

• 优点：代码简单，事务由数据库保证，真正的 ACID 原子性
• 缺点：单线程执行，CPU 密集的数据预处理无法并行
• 适用：绝大多数业务场景，数据量 10 万以内

注意事项

1. **必须开启 rewriteBatchedStatements=true**：这是性能的关键开关，不开的话 BATCH 模式只是减少了 MyBatis 层的开销，SQL 仍然逐条发送，实测差距可达 10 倍以上
2. flushStatements() 的间隔不宜过大：建议每 500~2000 条 flush 一次，过大会导致内存中积压大量 Statement 对象，有 OOM 风险
3. 长事务风险：单事务插入 10 万条以上时，事务持续时间可能超过几十秒，会导致 undo log 膨胀、锁持有时间过长、主从延迟加剧。如果出现这种情况建议切换到方案四
4. BATCH 模式下无法获取自增 ID 返回值：ExecutorType.BATCH 模式下 insert() 返回值不可靠（始终为 -2147482646），如果需要获取插入后的自增 ID，需要在 flush 后额外查询
5. 不要在 BATCH session 中混用查询：BATCH 模式的 SqlSession 中执行 SELECT 会触发隐式 flush，破坏批量攒 SQL 的效果

---

方案二：多线程准备 + 单线程提交

核心思路

如果数据需要复杂的校验、转换（CPU 密集型操作），可以 并行处理数据，串行入库。利用多线程加速数据准备阶段，最终仍然在单事务内完成插入。

示例代码

@Service
@RequiredArgsConstructor
public class ParallelPrepareBatchService {

    private final SqlSessionFactory sqlSessionFactory;
    private final ThreadPoolExecutor executor;

    @Transactional(rollbackFor = Exception.class)
    public void batchInsert(List<OrderItem> rawList) {
        // 阶段一：多线程并行做数据校验、转换、计算
        List<List<OrderItem>> partitions = ListUtil.partition(rawList, 500);
        List<CompletableFuture<List<OrderItem>>> futures = partitions.stream()
                .map(chunk -> CompletableFuture.supplyAsync(
                        () -> chunk.stream()
                                .map(this::validateAndEnrich)
                                .toList(),
                        executor))
                .toList();

        List<OrderItem> allEntities = futures.stream()
                .map(CompletableFuture::join)
                .flatMap(List::stream)
                .toList();

        // 阶段二：单事务批量插入
        try (SqlSession session = sqlSessionFactory.openSession(ExecutorType.BATCH)) {
            OrderItemMapper mapper = session.getMapper(OrderItemMapper.class);
            for (int i = 0; i < allEntities.size(); i++) {
                mapper.insert(allEntities.get(i));
                if ((i + 1) % 1000 == 0) {
                    session.flushStatements();
                }
            }
            session.flushStatements();
        }
    }

    private OrderItem validateAndEnrich(OrderItem item) {
        // 模拟 CPU 密集的数据校验和转换逻辑
        item.setTotalPrice(item.getPrice().multiply(BigDecimal.valueOf(item.getQuantity())));
        return item;
    }
}

优缺点

• 优点：数据准备阶段充分利用多核 CPU，插入阶段仍保持强一致事务
• 缺点：如果瓶颈在数据库 IO 而非 CPU，多线程准备意义不大
• 适用：数据校验/转换逻辑复杂、CPU 密集的场景

注意事项

1. 线程池要用自定义的有界线程池：绝对不要用 Executors.newFixedThreadPool() 等无界队列线程池，防止数据量暴涨时任务堆积导致 OOM。建议用 ThreadPoolExecutor 显式指定队列容量和拒绝策略
2. CompletableFuture.join() 会吞掉异常：如果子线程抛出异常，join() 会包装成 CompletionException 抛出，但如果你用了 thenApply 链式调用而没有正确传播异常，可能会导致”静默失败”。建议在 join 之后显式检查结果
3. 数据准备阶段不要有数据库操作：多线程准备阶段应该是纯 CPU 计算（校验、转换、计算），一旦涉及数据库查询就会引入连接争用，反而可能比单线程更慢
4. 内存占用翻倍：原始数据 + 处理后的数据同时存在于内存中，数据量大时注意 JVM 堆内存是否够用
5. 事务超时设置：如果数据准备耗时较长，@Transactional 的默认超时可能不够。准备阶段虽然不涉及数据库，但事务已经开启，时间会被计入事务超时。可以考虑把事务注解只加在实际插入的方法上

---

方案三：Staging 表伪 2PC

核心思路

先让多线程并发写入 staging（临时表），各自独立事务提交。全部完成后，协调者在 一个最终事务 里执行 INSERT INTO target SELECT FROM staging，对业务表而言仍然是原子的。

把问题拆成两层：

• 并发层：多线程写 staging，各自独立事务，互不干扰
• 原子层：最终一个事务做数据搬运，真正的数据库级原子性

示例代码

@Service
@RequiredArgsConstructor
public class StagingTwoPhaseService {

    private final StagingOrderItemMapper stagingMapper;
    private final OrderItemMapper orderItemMapper;
    private final BatchRecordMapper batchRecordMapper;
    private final ThreadPoolExecutor executor;
    private final PlatformTransactionManager transactionManager;

    public void batchInsert(List<OrderItem> dataList) {
        String batchId = IdUtil.fastSimpleUUID();
        int expectedCount = dataList.size();

        // 记录批次
        batchRecordMapper.insert(new BatchRecord(batchId, expectedCount, "STAGING"));

        // 阶段一：多线程并发写 staging 表
        List<List<OrderItem>> partitions = ListUtil.partition(dataList, 500);
        AtomicBoolean hasFailed = new AtomicBoolean(false);

        List<CompletableFuture<Void>> futures = partitions.stream()
                .map(chunk -> CompletableFuture.runAsync(() -> {
                    TransactionStatus status = transactionManager.getTransaction(
                            new DefaultTransactionDefinition());
                    try {
                        for (OrderItem item : chunk) {
                            StagingOrderItem staging = toStaging(item, batchId);
                            stagingMapper.insert(staging);
                        }
                        transactionManager.commit(status);
                    } catch (Exception e) {
                        hasFailed.set(true);
                        transactionManager.rollback(status);
                    }
                }, executor))
                .toList();

        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();

        // 检查是否有失败
        if (hasFailed.get()) {
            stagingMapper.deleteByBatchId(batchId);
            batchRecordMapper.updateStatus(batchId, "FAILED");
            throw new BusinessException("Staging 阶段有线程失败，已清理");
        }

        // 校验数据完整性
        long actualCount = stagingMapper.countByBatchId(batchId);
        if (actualCount != expectedCount) {
            stagingMapper.deleteByBatchId(batchId);
            batchRecordMapper.updateStatus(batchId, "FAILED");
            throw new BusinessException("数据完整性校验失败");
        }

        // 阶段二：单事务搬运到目标表（原子操作）
        batchRecordMapper.updateStatus(batchId, "COMMITTING");
        orderItemMapper.insertFromStaging(batchId);
        stagingMapper.deleteByBatchId(batchId);
        batchRecordMapper.updateStatus(batchId, "DONE");
    }
}

优缺点

• 优点：并发写入能力强，最终业务表仍满足原子性
• 缺点：需要 staging 表、批次状态表、清理机制；多一步数据搬运
• 适用：并发必须保留，但原子性必须严格保证在最终业务表

注意事项

1. staging 表不要加业务约束：staging 表不应有唯一索引、外键等约束，让并发写入尽量无冲突。所有业务校验应该在协调者校验阶段完成，而不是依赖数据库约束
2. 数据库连接池要够大：多线程并发写入时，每个线程占用一个数据库连接。如果线程数 > 连接池 max-active，会出现等待甚至死锁。确保 Druid.max-active ≥ 并发线程数 + 主线程预留
3. 协调者崩溃的数据残留：如果协调者在 staging 写入完成后、搬运之前崩溃，staging 中会残留”孤儿数据”。必须配合定时任务扫描 batch_record 中长时间处于 STAGING/COMMITTING 状态的批次，执行超时清理
4. INSERT INTO ... SELECT 的锁行为：在 InnoDB 中 INSERT INTO target SELECT FROM staging WHERE ... 会对 staging 表加共享锁（在 RR 隔离级别下还会加 gap lock）。如果此时恰好有其他批次也在写 staging，可能出现锁等待。建议 staging 表使用 READ COMMITTED 隔离级别或按 batch_id 做表分区
5. 大数据量的搬运瓶颈：INSERT ... SELECT 虽然是数据库内部搬运，但如果数据量很大（比如 50 万行），这一步本身也是一个大事务，和方案一面临一样的长事务风险
6. 事务传播要注意：commitToTarget() 方法上的 @Transactional 要确保是 REQUIRES_NEW 或者由顶层方法统一管理，避免嵌套事务导致提交时机不符预期

---

方案四：分批提交 + 补偿机制

核心思路

当数据量大到单事务不现实时（长事务锁资源、undo log 膨胀、主从延迟），放弃强原子性，转向 最终一致性。每批独立事务提交，失败时根据 batch_id 做补偿回滚。

示例代码

@Service
@RequiredArgsConstructor
public class CompensateBatchService {

    private final SqlSessionFactory sqlSessionFactory;
    private final OrderItemMapper orderItemMapper;
    private final BatchRecordMapper batchRecordMapper;

    @Transactional(rollbackFor = Exception.class)
    public void insertChunk(String batchId, int chunkIndex, List<OrderItem> chunk) {
        try (SqlSession session = sqlSessionFactory.openSession(ExecutorType.BATCH)) {
            OrderItemMapper mapper = session.getMapper(OrderItemMapper.class);
            for (OrderItem item : chunk) {
                item.setBatchId(batchId);
                mapper.insert(item);
            }
            session.flushStatements();
        }
    }

    public void batchInsert(List<OrderItem> dataList) {
        String batchId = IdUtil.fastSimpleUUID();
        List<List<OrderItem>> chunks = ListUtil.partition(dataList, 2000);

        batchRecordMapper.insert(new BatchRecord(batchId, dataList.size(), "PROCESSING"));

        for (int i = 0; i < chunks.size(); i++) {
            try {
                insertChunk(batchId, i, chunks.get(i));
            } catch (Exception e) {
                // 触发补偿：回滚已成功的 chunk
                compensate(batchId);
                batchRecordMapper.updateStatus(batchId, "ROLLED_BACK");
                throw new BusinessException("批量插入失败，已触发补偿回滚: " + e.getMessage());
            }
        }

        batchRecordMapper.updateStatus(batchId, "COMPLETED");
    }

    private void compensate(String batchId) {
        orderItemMapper.deleteByBatchId(batchId);
    }
}

优缺点

• 优点：无长事务风险，适合超大数据量，性能最高
• 缺点：补偿回滚前有短暂中间状态，设计复杂度高
• 适用：百万级数据量，可接受最终一致性

注意事项

1. 补偿操作本身也可能失败：DELETE FROM order_item WHERE batch_id = ? 如果因为网络抖动、数据库宕机等原因失败，就会出现”既没全写完，也没全删干净”的状态。补偿逻辑必须支持重试，建议配合定时任务扫描 ROLLED_BACK 状态的批次，确认数据确实已清理
2. 补偿窗口期的脏读：在第 3 批失败、触发补偿删除第 1~2 批数据的这个窗口期内，其他业务如果查询了 order_item 表，会读到”不完整的数据”。解决方案：业务查询加上 status = 'COMPLETED' 的批次过滤，或者在数据上增加一个 visible 标志位，最后一步统一设为可见
3. batch_id 字段必须加索引：补偿删除 DELETE WHERE batch_id = ? 如果没有索引，会全表扫描加锁，在数据量大时可能导致严重的锁冲突。init.sql 中已为此字段建立了索引
4. 分批大小的选择：每批太小（如 100 条）会导致事务次数过多、频繁提交的开销大于批量的收益；每批太大（如 10 万条）又回到了长事务的问题。建议根据实际压测调整，一般 1000~5000 条为宜
5. 幂等性设计：如果补偿后需要重试整个批次，要确保插入操作是幂等的。可以用 INSERT IGNORE 或 ON DUPLICATE KEY UPDATE，或者每次重试生成新的 batch_id
6. 并发场景可结合多线程：方案四本身是串行分批，但也可以改成多线程并发分批插入（每个线程处理若干 chunk），补偿逻辑不变。代价是补偿时需要处理更复杂的并发状态

---

关于 Seata 分布式事务

如果业务场景是 跨库/跨服务，需要考虑 Seata 等分布式事务框架：

AT 模式 vs XA 模式

维度	AT 模式	XA 模式
一致性级别	最终一致（有短暂中间状态）	强一致
隔离性	分支事务提交后可被读到	prepare 阶段数据对外不可见
性能	高（立即释放锁）	低（prepare 到 commit 一直持锁）
回滚机制	反向 SQL 补偿（undo_log）	数据库原生 XA ROLLBACK
部署要求	需要 Seata Server + undo_log 表	需要 Seata Server + 数据库支持 XA

什么时候该用 Seata？

• 单库批量插入 → 不需要，本文方案就够
• 跨库/跨服务的写入必须原子 → Seata 是正解
• 多线程写同一库同一表 → 本质上不需要分布式事务

---

如何选择？

数据量 < 10万，单库单表？
    └─ YES → 方案一（单事务 BATCH）直接搞定
    └─ NO ↓

数据准备阶段 CPU 密集？
    └─ YES → 方案二（多线程准备 + 单线程提交）
    └─ NO ↓

必须并发写入，且要求业务表强原子？
    └─ YES → 方案三（Staging 伪 2PC）
    └─ NO ↓

数据量百万级，可接受最终一致？
    └─ YES → 方案四（分批 + 补偿）

跨库/跨服务？
    └─ Seata AT/XA 模式

实战建议：90% 的场景方案一就够用。在没有性能压测数据证明单事务是瓶颈之前，不要过早引入复杂方案。

---

完整示例项目

本文所有方案的完整可运行代码已集成到 Spring Boot 示例项目中，提供了统一的 REST API 接口方便测试对比。

POST /api/batch/single-tx          # 方案一：单事务 BATCH
POST /api/batch/parallel-prepare   # 方案二：多线程准备 + 单线程提交
POST /api/batch/staging-2pc        # 方案三：Staging 伪 2PC
POST /api/batch/compensate         # 方案四：分批 + 补偿
DELETE /api/batch/clean             # 清理所有测试数据