Ensuring Full Transaction Rollback in Multi‑Threaded Spring Boot MyBatis Operations

Background Introduction

In a scenario where a massive amount of data needs to be inserted into a database, additional modification operations must be performed first. To improve response time, the insert operation is split across multiple threads. If any thread fails, the entire operation should be rolled back.

Common Classes and Methods

Utility methods and configuration classes are provided to support the multithreaded transaction:

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

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

Example Transaction Failure

Using the @Transactional annotation on a method that spawns child threads does not roll back the delete operation performed in the main thread when a child thread throws an exception. The following code demonstrates the problem:

@Override
@Transactional
public void saveThread(List<EmployeeDO> employeeDOList) {
    try {
        // Delete operation that will not be rolled 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 (Thread t : threadArray) {
            service.execute(t);
        }
        countDownLatch.await();
        System.out.println("添加完毕");
    } catch (Exception e) {
        log.info("error", e);
        throw new ServiceException("002", "出现异常");
    } finally {
        connection.close();
    }
}

When one of the child threads throws an exception, the delete operation executed before spawning the threads remains committed, leading to inconsistent data.

Using SqlSession to Control Manual Commit

To achieve proper rollback, the transaction is managed manually by obtaining a SqlSession, disabling auto‑commit, and committing or rolling back based on the results of all child threads.

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

<!-- MyBatis mapper snippet -->
<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>

Test results show that when any child thread fails, the delete operation is rolled back and the database remains unchanged. In the successful case, both delete and insert operations are committed.

Database state after failure:

Database state after success: