Handling Multithreaded Transactions in Spring Using Manual SqlSession Management

Background

When inserting a large volume of data, the operation is split across multiple threads to improve response time, and if any thread fails, the whole transaction should roll back. The standard @Transactional annotation does not work across threads because the child threads do not share the same transaction context.

Common Utility Classes

/**
 * Average split list method.
 */
public static <T> List<List<T>> averageAssign(List<T> source, int n) {
    List<List<T>> result = new ArrayList<>();
    int remaider = source.size() % n;
    int number = source.size() / n;
    int offset = 0; // offset
    for (int i = 0; i < n; i++) {
        List<T> value = null;
        if (remaider > 0) {
            value = source.subList(i * number + offset, (i + 1) * number + offset + 1);
            remaider--;
            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() {}
}

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

Failed Multithreaded Transaction Example

@Override
@Transactional
public void saveThread(List<EmployeeDO> employeeDOList) {
    try {
        // delete operation in main thread (won't roll back)
        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();
    }
}

In this scenario, when a child thread throws an exception, the delete operation performed by the main thread does not roll back, demonstrating that @Transactional is ineffective across threads.

Solution Using Manual SqlSession Transaction Control

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

This approach obtains a SqlSession to control the transaction manually: set auto‑commit to false, execute delete and batch inserts in child threads via Callable, and commit only if all sub‑tasks succeed; otherwise, roll back the whole transaction.

Test results show that when an exception occurs in any child thread, both the delete and insert operations are rolled back, confirming successful transaction management.

Source: blog.csdn.net/weixin_43225491/article/details/117705686