How to Ensure Transaction Rollback in Spring Multithreaded Operations

Background Introduction

In a scenario where a large amount of data needs to be inserted into the database, additional modification operations are performed first, followed by the insert. Because the insert volume can be huge, the data is split and processed in parallel using multiple threads to improve response time. If any thread fails, the whole operation should roll back.

In Spring, the @Transactional annotation can control transactions so that an exception triggers a rollback. However, in a multithreaded context this annotation does not take effect: when a child thread throws an exception, the modifications made by the main thread are not rolled back, leading to data inconsistency.

The following sections present a simple example that demonstrates multithreaded transaction handling.

Common Utility Classes and Methods

/**
 * Average split list method.
 * @param source the original list
 * @param n      number of sub‑lists
 * @return a list containing n sub‑lists
 */
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; // offset
    for (int i = 0; i < n; i++) {
        List<T> value;
        if (remainder > 0) {
            value = source.subList(i * number + offset, (i + 1) * number + offset + 1);
            remainder--;
            offset++;
        } else {
            value = source.subList(i * number + offset, (i + 1) * number + offset);
        }
        result.add(value);
    }
    return result;
}

/** Thread pool configuration */
public class ExecutorConfig {
    private static int maxPoolSize = Runtime.getRuntime().availableProcessors();
    private volatile static ExecutorService executorService;
    public static ExecutorService getThreadPool() {
        if (executorService == null) {
            synchronized (ExecutorConfig.class) {
                if (executorService == null) {
                    executorService = newThreadPool();
                }
            }
        }
        return executorService;
    }
    private static ExecutorService newThreadPool() {
        int queueSize = 500;
        int corePool = Math.min(5, maxPoolSize);
        return new ThreadPoolExecutor(corePool, maxPoolSize, 10000L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(queueSize), new ThreadPoolExecutor.AbortPolicy());
    }
    private ExecutorConfig() {}
}

/** Obtain SqlSession */
@Component
public class SqlContext {
    @Resource
    private SqlSessionTemplate sqlSessionTemplate;
    public SqlSession getSqlSession() {
        SqlSessionFactory sqlSessionFactory = sqlSessionTemplate.getSqlSessionFactory();
        return sqlSessionFactory.openSession();
    }
}

Example of Transaction Failure

@Override
@Transactional
public void saveThread(List<EmployeeDO> employeeDOList) {
    try {
        // Delete first; this will not roll back if a child thread fails
        this.getBaseMapper().delete(null);
        ExecutorService service = ExecutorConfig.getThreadPool();
        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", "出现异常");
                    }
                    this.saveBatch(list);
                } finally {
                    countDownLatch.countDown();
                }
            });
        }
        for (int i = 0; i < lists.size(); i++) {
            service.execute(threadArray[i]);
        }
        countDownLatch.await();
        System.out.println("添加完毕");
    } catch (Exception e) {
        log.info("error", e);
        throw new ServiceException("002", "出现异常");
    } finally {
        connection.close();
    }
}

The test shows that when one child thread throws an exception, other threads also report failure, but the delete operation performed by the main thread does not roll back, indicating that @Transactional is ineffective in this multithreaded context.

Solution Using Manual SqlSession Commit

@Override
public void saveThread(List<EmployeeDO> employeeDOList) throws SQLException {
    SqlSession sqlSession = sqlContext.getSqlSession();
    Connection connection = sqlSession.getConnection();
    try {
        connection.setAutoCommit(false);
        EmployeeMapper employeeMapper = sqlSession.getMapper(EmployeeMapper.class);
        employeeMapper.delete(null);
        ExecutorService service = ExecutorConfig.getThreadPool();
        List<Callable<Integer>> callableList = new ArrayList<>();
        List<List<EmployeeDO>> lists = averageAssign(employeeDOList, 5);
        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);
            Callable<Integer> callable = () -> {
                if (!atomicBoolean.get()) {
                    throw new ServiceException("001", "出现异常");
                }
                return employeeMapper.saveBatch(list);
            };
            callableList.add(callable);
        }
        List<Future<Integer>> futures = service.invokeAll(callableList);
        for (Future<Integer> future : futures) {
            if (future.get() <= 0) {
                connection.rollback();
                return;
            }
        }
        connection.commit();
        System.out.println("添加完毕");
    } catch (Exception e) {
        connection.rollback();
        log.info("error", e);
        throw new ServiceException("002", "出现异常");
    } finally {
        connection.close();
    }
}

By obtaining a SqlSession and manually controlling the transaction (setting auto‑commit to false, committing on success, rolling back on any failure), the delete operation is also rolled back when a child thread fails, achieving the desired atomicity.

Successful Operation Example

@Override
public void saveThread(List<EmployeeDO> employeeDOList) throws SQLException {
    SqlSession sqlSession = sqlContext.getSqlSession();
    Connection connection = sqlSession.getConnection();
    try {
        connection.setAutoCommit(false);
        EmployeeMapper employeeMapper = sqlSession.getMapper(EmployeeMapper.class);
        employeeMapper.delete(null);
        ExecutorService service = ExecutorConfig.getThreadPool();
        List<Callable<Integer>> callableList = new ArrayList<>();
        List<List<EmployeeDO>> lists = averageAssign(employeeDOList, 5);
        for (int i = 0; i < lists.size(); i++) {
            List<EmployeeDO> list = lists.get(i);
            Callable<Integer> callable = () -> employeeMapper.saveBatch(list);
            callableList.add(callable);
        }
        List<Future<Integer>> futures = service.invokeAll(callableList);
        for (Future<Integer> future : futures) {
            if (future.get() <= 0) {
                connection.rollback();
                return;
            }
        }
        connection.commit();
        System.out.println("添加完毕");
    } catch (Exception e) {
        connection.rollback();
        log.info("error", e);
        throw new ServiceException("002", "出现异常");
    } finally {
        connection.close();
    }
}

Test results confirm that the delete operation is rolled back and the insert succeeds, demonstrating that manual transaction control works correctly in a multithreaded environment.