在单体项目的大数据量入库场景中，为提升处理性能，开发者常采用“主线程前置操作+多线程拆分插入”的方案。但实际开发中，很多人会踩上Spring @Transactional注解的坑——子线程执行失败时，主线程的修改操作无法回滚，导致数据一致性被破坏。本文将深入剖析问题本质，通过完整的失败案例与优化后的解决方案，带你彻底解决多线程事务一致性难题。

一、背景与核心问题

1. 应用场景

需要处理大批量数据入库任务，流程如下：

1. 主线程先执行数据库前置修改（如清理历史数据）；

2. 按固定大小拆分全量数据为多个子列表（分片）；

3. 多线程并行执行批量插入，提升处理效率；

4. 核心要求：任一子线程执行失败，所有操作（主线程前置修改+子线程插入）需完全回滚，保证“要么全成功，要么全失败”。

2. 核心问题：@Transactional多线程失效

Spring的@Transactional注解基于线程本地事务管理机制（ThreadLocal），事务上下文与当前线程强绑定。在多线程场景中，子线程无法继承主线程的事务状态，导致：

• 子线程抛出的异常无法传播到主线程；

• 主线程的事务不会因子线程异常触发回滚；

• 最终出现“主线程修改已生效、子线程插入部分失败”的不一致状态。

二、典型失败案例汇总

失败案例1：@Transactional+多线程，子线程异常无法回滚主线程操作（核心场景）

代码实现（基于Spring Boot+MyBatis Plus）

@Service
public class EmployeeBOImpl implements EmployeeBO {

    @Autowired
    private EmployeeMapper employeeMapper;

    @Autowired
    private ExecutorService executorService;

    // ❌ @Transactional无法跨线程生效
    @Override
    @Transactional
    public void saveThread(List<EmployeeDO> employeeDOList) {
        try {
            // 主线程：删除历史数据（执行成功后若子线程异常，无法回滚）
            employeeMapper.delete(null);
            System.out.println("主线程：历史数据删除完成");

            // 数据拆分：按5份拆分列表
            List<List<EmployeeDO>> lists = averageAssign(employeeDOList, 5);
            Thread[] threadArray = new Thread[lists.size()];
            CountDownLatch countDownLatch = new CountDownLatch(lists.size());
            AtomicBoolean atomicBoolean = new AtomicBoolean(true);

            // 多线程执行插入
            for (int i = 0; i < lists.size(); i++) {
                // 让最后一个子线程抛出异常
                if (i == lists.size() - 1) {
                    atomicBoolean.set(false);
                }
                List<EmployeeDO> list = lists.get(i);
                threadArray[i] = new Thread(() -> {
                    try {
                        if (!atomicBoolean.get()) {
                            throw new ServiceException("001", "子线程执行失败");
                        }
                        // MyBatis Plus批量插入
                        this.saveBatch(list);
                    } finally {
                        countDownLatch.countDown();
                    }
                });
                executorService.execute(threadArray[i]);
            }

            // 等待所有子线程执行完毕
            countDownLatch.await();
            System.out.println("所有子线程执行完毕");
        } catch (Exception e) {
            log.info("执行异常", e);
            throw new ServiceException("002", "批量插入失败");
        }
    }

    // 列表平均拆分工具方法
    public static <T> List<List<T>> averageAssign(List<T> source, int n) {
        List<List<T>> result = new ArrayList<>();
        int remainder = source.size() % n; // 余数
        int number = source.size() / n;    // 每份基础数量
        int offset = 0; // 偏移量
        for (int i = 0; i < n; i++) {
            List<T> subList;
            if (remainder > 0) {
                subList = source.subList(i * number + offset, (i + 1) * number + offset + 1);
                remainder--;
                offset++;
            } else {
                subList = source.subList(i * number + offset, (i + 1) * number + offset);
            }
            result.add(subList);
        }
        return result;
    }
}

测试结果

• 子线程抛出ServiceException（code=001）；

• 主线程的删除操作未回滚，数据库中历史数据被清空，新数据仅部分插入（未抛出异常的子线程执行成功）；

• 最终数据处于“无历史数据、新数据不完整”的不一致状态。

失败原因

@Transactional的事务上下文存储在ThreadLocal中，子线程无法继承主线程的事务状态。子线程的异常仅终止自身执行，不会触发主线程事务的回滚机制。

失败案例2：共享连接但未关闭自动提交，部分操作即时生效

代码缺陷

手动获取数据库连接后，未设置connection.setAutoCommit(false)，导致子线程执行executeBatch()时即时提交，后续回滚无效。

// ❌ 关键错误：未关闭自动提交
connection = sqlSessionTemplate.getSqlSessionFactory().openSession().getConnection();
// 未执行 connection.setAutoCommit(false);

// 子线程执行批量插入时，操作即时提交
private void insertChunk(List<Data> chunk, Connection connection) throws SQLException {
    String sql = "INSERT INTO data_table (id, name) VALUES (?, ?)";
    try (PreparedStatement ps = connection.prepareStatement(sql)) {
        for (Data data : chunk) {
            ps.setLong(1, data.getId());
            ps.setString(2, data.getName());
            ps.addBatch();
        }
        ps.executeBatch(); // 自动提交，无法回滚
    }
}

失败原因

数据库连接默认开启自动提交（AutoCommit=true），子线程的批量插入操作会即时生效，即使后续子线程失败，已提交的操作也无法通过rollback()撤销。

失败案例3：子线程创建新连接，脱离全局事务控制

代码缺陷

主线程获取连接并开启手动事务，但子线程通过JdbcTemplate或新SqlSession创建独立连接，导致操作与主线程事务隔离。

@Autowired
private JdbcTemplate jdbcTemplate; // 子线程使用新连接

// 子线程执行逻辑（错误示例）
futures.add(executorService.submit(() -> {
    // ❌ 新连接：未复用主线程的事务连接
    String sql = "INSERT INTO data_table (id, name) VALUES (?, ?)";
    jdbcTemplate.batchUpdate(sql, new BatchPreparedStatementSetter() {
        @Override
        public void setValues(PreparedStatement ps, int i) throws SQLException {
            Data data = chunk.get(i);
            ps.setLong(1, data.getId());
            ps.setString(2, data.getName());
        }

        @Override
        public int getBatchSize() {
            return chunk.size();
        }
    });
    return true;
}));

失败原因

子线程的新连接默认开启自动提交，其操作与主线程的事务完全隔离，主线程的rollback()无法影响子线程的已提交操作。

失败案例4：Future.get()未设超时，导致事务挂起与连接泄漏

代码缺陷

Future.get()未设置超时时间，子线程执行超时（如数据库锁等待）时，主线程长期阻塞，连接无法释放。

// ❌ 无超时设置，主线程无限阻塞
for (Future<Boolean> future : futures) {
    future.get(); // 子线程阻塞时，主线程一直等待
}

失败原因

数据库连接长期处于“事务未提交/回滚”状态，既导致连接池资源泄漏，又可能引发数据库锁表，影响其他业务执行。

三、解决方案：手动事务控制+多线程协调

核心设计思路

要实现多线程事务一致性，需突破@Transactional的线程绑定限制，核心是共享数据库连接+手动控制事务生命周期，同时规避连接泄漏、超时挂起等生产隐患：

1. 所有线程复用同一个由Spring管理的数据库连接，避免SqlSession/连接泄漏；

2. 关闭自动提交，由业务逻辑显式控制commit()/rollback()，且关闭连接前恢复自动提交默认状态；

3. 通过Future监控所有子线程执行结果，设置超时时间，任一失败则全局回滚；

4. 规范日志打印与边界校验，提升代码健壮性与可排查性。

正确代码示例

1. 核心批量插入服务

import org.mybatis.spring.SqlSessionTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * 多线程批量插入服务（生产环境可用，无连接泄漏/事务挂起隐患）
 */
@Service
public class BatchInsertService {
    // 统一使用SLF4J日志框架，替代System.out.println，支持生产环境日志分级配置
    private static final Logger log = LoggerFactory.getLogger(BatchInsertService.class);

    @Autowired
    private SqlSessionTemplate sqlSessionTemplate;

    // 直接获取单例线程池，规避Spring容器注入失败问题
    private final ExecutorService executorService = ExecutorConfig.getThreadPool();

    /**
     * 多线程批量插入，保证全局事务一致性
     * @param dataList 待插入全量数据
     */
    public void batchInsertWithTransaction(List<Data> dataList) {
        Connection connection = null;
        boolean originalAutoCommit = true; // 记录连接原始AutoCommit状态，用于后续恢复

        try {
            // 1. 获取Spring管理的数据库连接（避免SqlSession/连接泄漏），开启手动事务
            connection = sqlSessionTemplate.getConnection();
            originalAutoCommit = connection.getAutoCommit(); // 保存原始自动提交状态
            connection.setAutoCommit(false); // 关闭自动提交，接管事务生命周期

            // 2. 主线程执行前置操作（如删除历史数据，纳入全局事务）
            deleteOldData(connection);

            // 3. 数据分片（按1000条/片，平衡性能与资源消耗）
            List<List<Data>> chunks = splitList(dataList, 1000);
            List<Future<Boolean>> futures = new ArrayList<>();

            // 4. 多线程并行插入（所有子线程复用同一个数据库连接，保证事务原子性）
            for (List<Data> chunk : chunks) {
                futures.add(executorService.submit(() -> {
                    try {
                        insertChunk(chunk, connection);
                        return true; // 分片插入成功返回true
                    } catch (Exception e) {
                        log.error("子线程分片插入失败，分片数据量：{}", chunk.size(), e);
                        return false; // 分片插入失败返回false
                    }
                }));
            }

            // 5. 监控所有子线程结果，任一失败/超时则全局回滚（设置30秒超时，避免事务挂起）
            for (Future<Boolean> future : futures) {
                try {
                    if (!future.get(30, TimeUnit.SECONDS)) {
                        connection.rollback();
                        log.error("子线程执行失败，触发全局事务回滚");
                        throw new RuntimeException("子线程执行失败，全局事务已回滚");
                    }
                } catch (TimeoutException e) {
                    connection.rollback();
                    log.error("子线程执行超时（30秒），触发全局事务回滚", e);
                    throw new RuntimeException("子线程执行超时，全局事务已回滚", e);
                }
            }

            // 6. 所有子线程执行成功，提交全局事务
            connection.commit();
            log.info("多线程批量插入成功，全量数据量：{}，事务已提交", dataList.size());

        } catch (Exception e) {
            // 7. 捕获任何异常，执行全局回滚，保证数据一致性
            if (connection != null) {
                try {
                    if (!connection.isClosed()) {
                        connection.rollback();
                        log.error("事务执行异常，已触发全局回滚", e);
                    }
                } catch (SQLException ex) {
                    log.error("事务回滚操作失败", ex);
                }
            }
            throw new RuntimeException("批量插入事务执行失败", e);
        } finally {
            // 8. 释放资源：关闭连接+恢复原始AutoCommit状态，避免连接池复用隐患
            if (connection != null) {
                try {
                    if (!connection.isClosed()) {
                        connection.setAutoCommit(originalAutoCommit); // 恢复连接原始配置
                        connection.close();
                        log.info("数据库连接已关闭，且恢复AutoCommit原始状态：{}", originalAutoCommit);
                    }
                } catch (SQLException e) {
                    log.error("数据库连接关闭失败", e);
                }
            }
        }
    }

    /**
     * 数据分片工具：按固定大小拆分列表，规避边界异常
     * @param list 全量数据
     * @param chunkSize 每片最大容量（如1000）
     * @return 分片后的子列表集合
     */
    private List<List<Data>> splitList(List<Data> list, int chunkSize) {
        // 边界校验：避免分片大小非法或数据为空导致的异常
        if (chunkSize <= 0) {
            throw new IllegalArgumentException("分片大小必须大于0");
        }
        if (list == null || list.isEmpty()) {
            log.warn("待分片数据列表为空，无需执行分片操作");
            return new ArrayList<>();
        }

        List<List<Data>> chunks = new ArrayList<>();
        for (int i = 0; i < list.size(); i += chunkSize) {
            int endIndex = Math.min(i + chunkSize, list.size());
            chunks.add(list.subList(i, endIndex));
        }
        return chunks;
    }

    /**
     * 主线程前置操作：删除历史数据（共享全局事务连接）
     * @param connection 全局事务连接
     * @throws SQLException SQL执行异常
     */
    private void deleteOldData(Connection connection) throws SQLException {
        String deleteSql = "DELETE FROM data_table";
        try (PreparedStatement ps = connection.prepareStatement(deleteSql)) {
            int deleteCount = ps.executeUpdate();
            log.info("历史数据删除完成（纳入全局事务），删除记录数：{}", deleteCount);
        }
    }

    /**
     * 子线程批量插入：分片执行（共享全局事务连接，批量操作提升性能）
     * @param chunk 单批次数据
     * @param connection 全局事务连接
     * @throws SQLException SQL执行异常
     */
    private void insertChunk(List<Data> chunk, Connection connection) throws SQLException {
        if (chunk == null || chunk.isEmpty()) {
            return;
        }

        String insertSql = "INSERT INTO data_table (id, name) VALUES (?, ?)";
        try (PreparedStatement ps = connection.prepareStatement(insertSql)) {
            for (Data data : chunk) {
                ps.setLong(1, data.getId());
                ps.setString(2, data.getName());
                ps.addBatch(); // 批量添加SQL，减少网络IO交互
            }
            int[] batchResult = ps.executeBatch(); // 批量执行插入，返回每条记录执行结果
            log.info("单批次插入成功，分片数据量：{}，数据库执行结果条数：{}", chunk.size(), batchResult.length);
        }
    }
}

2. 线程池配置类（单例模式，适合生产环境）

import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 线程池配置类（单例模式，避免线程池泛滥，适配IO密集型任务）
 */
public class ExecutorConfig {
    // 核心参数：根据服务器CPU核心数动态调整，适配不同部署环境
    private static final int CPU_CORE_NUM = Runtime.getRuntime().availableProcessors();
    private static volatile ExecutorService executorService;

    /**
     * 双重检查锁创建单例线程池，保证全局唯一
     * @return 单例ExecutorService实例
     */
    public static ExecutorService getThreadPool() {
        if (executorService == null || executorService.isShutdown()) {
            synchronized (ExecutorConfig.class) {
                if (executorService == null || executorService.isShutdown()) {
                    executorService = newThreadPool();
                }
            }
        }
        return executorService;
    }

    /**
     * 构建IO密集型线程池（生产环境安全配置，避免任务丢失/内存溢出）
     */
    private static ExecutorService newThreadPool() {
        int corePoolSize = Math.max(2, CPU_CORE_NUM); // 核心线程数：至少2个，适配低配服务器
        int maxPoolSize = CPU_CORE_NUM * 2; // 最大线程数：CPU核心数*2（IO密集型任务最优配置）
        long keepAliveTime = 60L; // 空闲线程存活时间：60秒，释放闲置资源
        int queueCapacity = 1000; // 任务队列容量：1000，平衡任务堆积与内存消耗

        return new ThreadPoolExecutor(
                corePoolSize,
                maxPoolSize,
                keepAliveTime,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(queueCapacity),
                Thread::new, // 自定义线程创建器，便于后续线程问题排查
                new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略：调用者运行，避免关键任务丢失
        );
    }

    // 私有构造方法：禁止外部实例化，保证单例特性
    private ExecutorConfig() {}
}

3. Data实体类（对应数据库表）

/**
 * 数据实体类（对应数据库data_table表，支持MyBatis/数据库操作字段映射）
 */
public class Data {
    private Long id; // 主键ID（对应数据库表id字段）
    private String name; // 名称字段（对应数据库表name字段）

    // 无参构造方法：MyBatis与数据库操作必需
    public Data() {}

    // 有参构造方法：便于测试场景快速创建实例
    public Data(Long id, String name) {
        this.id = id;
        this.name = name;
    }

    // getter/setter方法：保证字段映射与数据读写正常
    public Long getId() {
        return id;
    }

    public void setId(Long id) {
        this.id = id;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }
}

测试结果验证

1. 子线程异常场景：任一子线程插入失败，future.get()返回false，立即触发connection.rollback()，主线程删除操作与所有子线程插入操作全部回滚，数据库数据保持初始一致状态；

2. 子线程超时场景：子线程执行超过30秒，触发TimeoutException，快速执行全局回滚并关闭连接，避免事务挂起与连接泄漏；

3. 全部成功场景：所有子线程分片插入完成，无异常/超时，connection.commit()提交全局事务，历史数据被清理，新数据全部入库，数据状态完整；

4. 连接复用场景：连接关闭前恢复AutoCommit原始状态，后续连接池复用该连接时，不会影响其他业务的事务配置，无未知隐患。

四、数据分片逻辑深度解析

多线程批量插入的性能瓶颈之一是数据分片策略，合理的分片能平衡线程负载与数据库压力，同时降低异常风险。

1. 核心分片策略：按数据量等量切分

实现逻辑

按固定大小（如1000条）将全量数据拆分为多个子列表，通过Math.min()处理最后一个子列表的边界，避免索引越界，且保证每个子列表数据量均匀。

选择依据（对比其他维度）

分片维度	优势	劣势
数据量	1. 控制单次插入规模，避免SQL语句过长超出网络/数据库限制； 2. 分片均衡，线程负载均匀，无部分线程过载； 3. 实现简单，无额外预处理开销，执行效率高； 4. 通用性强，适配绝大多数无特殊业务关联的批量插入场景	无明显致命劣势，是批量插入场景的最优解
业务属性（如类型）	同类数据集中处理，便于后续业务追溯与关联操作	1. 数据分布不均易导致线程负载失衡； 2. 依赖业务数据特性，通用性差； 3. 需额外解析业务字段，增加处理开销
时间戳	适合时序数据（如日志、流水）的批量处理场景	1. 数据易集中在某一时间段，导致分片不均衡； 2. 需额外对数据排序，增加处理复杂度与性能开销； 3. 超时风险更高，不利于事务控制

2. 分片大小（1000条）的选择理由

1. 数据库性能：单次批量插入1000-2000条数据时，数据库执行效率最优，过多会导致锁表时间过长，阻塞其他业务操作，过少则会增加网络IO交互次数，降低整体性能；

2. 网络传输：避免单次SQL语句过大，超出数据库数据包大小限制或网络传输MTU限制，导致请求被拒绝或传输超时；

3. 内存占用：控制单个线程处理的数据量，避免线程内加载过多数据导致JVM内存溢出（OOM），尤其在数据字段较多、单条数据体积较大的场景；

4. 超时控制：单次分片操作可在合理时间内完成（配合30秒全局超时），便于后续监控、告警与重试机制的落地。

3. 分片大小调整建议

• 数据库性能优异（如专用数据库集群、高配置服务器）：可将分片大小调整为2000-5000条，进一步提升批量处理效率；

• 单条数据体积大（如含大文本、Blob字段）：可将分片大小减小至500条以内，控制内存占用与单次SQL执行时间；

• 生产环境最终最优值：建议通过压测确定，结合数据库负载、网络延迟、服务器内存等实际场景综合评估，避免仅凭经验值配置。

4. 可选分片策略（扩展）

（1）按业务字段哈希分片

适合需同类数据集中处理的场景（如按用户类型、地区分片），保证同类数据在同一个线程中执行，便于业务关联与后续排查：

Map<String, List<Data>> chunksByType = dataList.stream()
        .collect(Collectors.groupingBy(Data::getType));

（2）按ID范围分片

适合ID连续递增的数据（如自增主键、有序编号），分片均衡性更优，且便于后续按ID范围追溯执行结果：

int total = dataList.size();
int threadNum = 10; // 预设线程数，与线程池最大线程数匹配
int perThreadSize = total / threadNum;
List<List<Data>> chunks = new ArrayList<>();
for (int i = 0; i < threadNum; i++) {
    int startIndex = i * perThreadSize;
    int endIndex = (i == threadNum - 1) ? total : (i + 1) * perThreadSize;
    chunks.add(dataList.subList(startIndex, endIndex));
}

五、关键注意事项与技术对比

1. 核心注意点（生产环境必看）

• 共享连接是事务一致性的前提：所有线程必须复用同一个Connection对象，否则无法保证操作纳入同一个全局事务，回滚机制失效；

• 连接生命周期闭环管理：必须在finally块中关闭连接，且恢复AutoCommit原始状态，避免连接泄漏与连接池复用隐患；

• 超时与异常全覆盖：Future.get()必须设置超时时间，且捕获TimeoutException，同时所有数据库操作异常需向上传播，保证回滚逻辑触发；

• 性能优化优先级：批量插入优先使用addBatch()+executeBatch()，减少网络IO交互次数，这是提升批量插入性能的核心手段，优于单纯增加线程数；

• 线程池配置安全：避免使用无界队列与不合理拒绝策略，防止任务堆积导致内存溢出，关键业务推荐使用CallerRunsPolicy拒绝策略，避免任务丢失。

2. 与@Transactional的核心区别

特性	@Transactional注解	手动事务控制（本文方案）
事务上下文存储	线程本地存储（ThreadLocal）	共享数据库连接（跨线程）
多线程场景支持	不支持（子线程无法继承事务状态）	支持（所有线程复用同一连接）
提交/回滚控制	自动控制（异常时自动回滚，无异常自动提交）	显式控制（代码调用commit/rollback，灵活度高）
连接管理	Spring容器自动管理，无需手动干预	手动管理（需关闭连接+恢复原始配置，要求更严谨）
超时控制	无内置超时，需额外配置全局事务超时	可自定义子线程超时，快速触发回滚，避免挂起
适用场景	单线程简单事务、无强一致性要求的场景	多线程批量处理、高并发、强一致性要求的场景
生产隐患风险	低（单线程场景）/ 高（多线程场景失效）	低（规范操作下，无连接泄漏/事务挂起隐患）

六、总结

多线程批量插入场景中，@Transactional因线程本地绑定的特性失效，无法保障跨线程的事务一致性，最终会导致数据混乱。本文提出的解决方案核心是手动管理数据库连接+显式控制事务生命周期+多线程结果监控：

1. 通过复用单个数据库连接，突破线程绑定限制，将所有操作纳入同一个全局事务；

2. 关闭自动提交，通过commit()/rollback()实现“要么全成功，要么全失败”的事务原子性；

3. 配置合理的线程池与分片策略，平衡处理性能与系统资源消耗；

4. 补充完整的异常与超时处理，恢复连接原始配置，规避生产环境的连接泄漏、事务挂起等隐患。

该方案适用于高并发大数据量批处理场景，虽然相比@Transactional增加了代码复杂度，但保证了数据的强一致性，且符合生产环境的部署要求。通过本文的失败案例分析与优化后的解决方案，你可以彻底解决多线程事务一致性难题，在提升数据处理性能的同时，保障数据安全可靠。