How to Ensure Transaction Rollback in Multithreaded Spring Applications

Background: In a scenario where a large amount of data needs to be inserted, some preliminary modifications are performed before the insertion. To improve response time, the data is split and processed in parallel using multiple threads. If any thread fails, the whole operation should roll back.

In Spring, the @Transactional annotation rolls back the transaction when an exception occurs, but it does not work across multiple threads. Consequently, when a child thread throws an exception, the modifications made by the main thread are not rolled back, leading to data inconsistency.

/**
 * Average split list method.
 */
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;
}

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() {}
}

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

Example of Transaction Failure

The following method uses @Transactional and creates child threads to insert data. The last thread deliberately throws an exception. Although the child threads fail, the delete operation performed before thread creation is not rolled back.

@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();
    }
}

Test results show that the child threads throw exceptions, but the delete operation remains, indicating that @Transactional did not roll back the whole transaction.

Solution: Use manual transaction control via SqlSession. Obtain the underlying JDBC connection, set auto‑commit to false, execute the delete, run child threads as Callable tasks, and decide to commit or roll back based on the results of all futures.

@Resource
SqlContext sqlContext;
@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();
    }
}

Tests of this approach show that when a child thread fails, the delete operation is rolled back and no data remains in the database. When all threads succeed, the delete is committed and the new records are inserted successfully.