How to Ensure Transaction Rollback in Multi‑Threaded Spring Applications

In a scenario where a large amount of data needs to be inserted into a database after some preliminary modifications, multithreading is used to split the data and improve response time, but if any thread fails, the entire operation must be rolled back.

Spring's @Transactional annotation does not work in child threads, so changes made by the main thread are not rolled back when a child thread throws an exception.

A simple example demonstrates multithreaded transaction handling.

Two solutions are provided: using a manual SqlSession with explicit commit/rollback, and using a thread pool with Callable tasks.

01 Common Utility Class and Methods

/**
 * Average split list method.
 * @param source the source list
 * @param n number of sublists
 * @param <T> element type
 * @return list of sublists
 */
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> value = null;
        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

/** Thread pool configuration */
public class ExecutorConfig {
    private static int maxPoolSize = Runtime.getRuntime().availableProcessors();
    private volatile static ExecutorService executorService;
    public static synchronized ExecutorService getThreadPool() {
        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() {}
}

SqlContext for Manual SqlSession

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

Unsuccessful Transaction Example (Transactional Annotation Fails)

@Transactional(rollbackFor = Exception.class)
public void saveThread(List<EmployeeDO> employeeDOList) {
    try {
        // Delete operation that will not roll back if child thread fails
        this.getBaseMapper().delete(null);
        ExecutorService service = ExecutorConfig.getThreadPool();
        List<List<EmployeeDO>> lists = averageAssign(employeeDOList, 2);
        Thread[] threadArray = new Thread[lists.size()];
        CountDownLatch countDownLatch = new CountDownLatch(lists.size());
        AtomicBoolean atomicBoolean = new AtomicBoolean(false);
        for (int i = 0; i < lists.size(); i++) {
            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 (Thread t : threadArray) {
            service.execute(t);
        }
        countDownLatch.await();
        System.out.println("添加完毕");
    } catch (Exception e) {
        log.info("error", e);
        throw new ServiceException("002", "出现异常");
    }
}

Test results show that when all child threads throw exceptions, the delete operation performed by the main thread is not rolled back, confirming that @Transactional does not take effect across threads.

Using SqlSession to Control Manual Transaction Commit

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 (List<EmployeeDO> list : lists) {
            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();
    }
}

MyBatis Mapper Insert Statement

<insert id="saveBatch" parameterType="List">
INSERT INTO employee (employee_id, age, employee_name, birth_date, gender, id_number, creat_time, update_time, status)
VALUES
<foreach collection="list" item="item" separator=",">
    (#{item.employeeId}, #{item.age}, #{item.employeeName}, #{item.birthDate}, #{item.gender}, #{item.idNumber}, #{item.creatTime}, #{item.updateTime}, #{item.status})
</foreach>
</insert>

Successful Transaction Example

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 (List<EmployeeDO> list : lists) {
            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();
    }
}

After applying the manual SqlSession approach, the delete operation is rolled back when an exception occurs, confirming that the transaction is successfully managed across multiple threads.