多线程JPA测试无法回滚的原因及解决方法

问题背景 #

在@DataJpaTest环境下，单线程测试的事务能正常回滚，但子线程执行的事务无法回滚，导致测试数据残留，影响后续测试用例的执行结果。

单线程测试代码及结果 #

@Test
void SingleThreadTest1() throws RequesterException, InterruptedException {
    System.out.println("(before)count: " + logRepository.count());
    App app = appRepository.findByName("TestApp").orElseThrow();
    myService.loggingTransaction(app);
    System.out.println("(after)count: " + logRepository.count());
}

@Test
void SingleThreadTest2() throws RequesterException, InterruptedException {
    System.out.println("(before)count: " + logRepository.count());
    App app = appRepository.findByName("TestApp").orElseThrow();
    myService.loggingTransaction(app);
    System.out.println("(after)count: " + logRepository.count());
}

执行结果：

#SingleThreadTest1
(before)count: 0
(after)count: 1

#SingleThreadTest2
(before)count: 0
(after)count: 1

多线程测试代码及结果 #

@DataJpaTest
class ServiceTest {
    static final int CONCURRENCY = 1;

    @Test
    void MultiThreadTest1() throws RequesterException, InterruptedException {
        System.out.println("(before)count: " + logRepository.count());
        App app = appRepository.findByName("TestApp").orElseThrow();
        // --------From Here: Spawn sub thread---------
        ThreadPoolExecutor executor = new ThreadPoolExecutor(CONCURRENCY, CONCURRENCY, 1, TimeUnit.MINUTES, new SynchronousQueue<>());
        executor.prestartAllCoreThreads();
        List<Callable<OptoutMessage>> callables = IntStream.range(0, CONCURRENCY)
                .<Callable<OptoutMessage>>mapToObj(i -> () -> myService.loggingTransaction(app))
                .toList();
        List<Future<OptoutMessage>> futures = executor.invokeAll(callables);
        executor.shutdownNow();
        // ------------------Until here----------------
        System.out.println("(after)count: " + logRepository.count());
    }
    
    @Test
    void MultiThreadTest2() throws RequesterException, InterruptedException {
        System.out.println("(before)count: " + logRepository.count());
        App app = appRepository.findByName("TestApp").orElseThrow();
        // --------From Here: Spawn sub thread---------
        ThreadPoolExecutor executor = new ThreadPoolExecutor(CONCURRENCY, CONCURRENCY, 1, TimeUnit.MINUTES, new SynchronousQueue<>());
        executor.prestartAllCoreThreads();
        List<Callable<OptoutMessage>> callables = IntStream.range(0, CONCURRENCY)
                .<Callable<OptoutMessage>>mapToObj(i -> () -> myService.loggingTransaction(app))
                .toList();
        List<Future<OptoutMessage>> futures = executor.invokeAll(callables);
        executor.shutdownNow();
        // ------------------Until here----------------
        System.out.println("(after)count: " + logRepository.count());
    }
}

执行结果：

#MultiThreadTest1
(before)count: 0
(after)count: 1

#MultiThreadTest2
(before)count: 1 // <- 此处应为0！！！
(after)count: 2 

核心问题 #

1. 为何@DataJpaTest中，子线程执行的事务无法回滚？
2. 在此场景下，如何实现@Test执行后的数据回滚？

问题解答 #

1. 子线程事务无法回滚的原因 #

@DataJpaTest默认会为每个测试方法绑定一个线程专属的事务上下文：Spring的事务管理器通过ThreadLocal存储当前线程的事务信息，只有测试方法所在主线程中执行的操作，才会被纳入这个事务，测试结束后Spring会自动回滚该主线程的事务。

而手动创建的子线程无法继承主线程的事务上下文，当myService.loggingTransaction在子线程中执行时，会触发全新的独立事务（如果方法标注了@Transactional，默认传播行为是REQUIRED，无现有事务则新建）。这个子线程事务不受Spring测试框架管控，执行完成后会自动提交，测试结束后Spring仅回滚主线程的空事务，子线程提交的数据自然残留。

2. 实现测试后数据回滚的可行方案 #

方案一：手动清理测试数据（推荐）

在每个测试方法结束后，直接删除子线程生成的数据，可通过@AfterEach注解统一执行清理逻辑：

@DataJpaTest
class ServiceTest {
    static final int CONCURRENCY = 1;
    @Autowired
    private LogRepository logRepository;

    @AfterEach
    void cleanUpTestData() {
        // 清空测试生成的日志数据
        logRepository.deleteAll();
    }

    // 多线程测试代码...
}

如果数据结构复杂，也可以用@Sql注解执行SQL脚本完成清理：

@Sql(scripts = "/cleanup-test-data.sql", executionPhase = Sql.ExecutionPhase.AFTER_TEST_METHOD)
@Test
void MultiThreadTest1() {
    // 测试逻辑...
}

方案二：让子线程共享主线程事务上下文（仅适用于简单场景）

通过Spring的TransactionSynchronizationManager手动将主线程的事务上下文传递给子线程，但需注意线程安全和事务冲突问题：

@Test
void MultiThreadTest1() throws RequesterException, InterruptedException {
    System.out.println("(before)count: " + logRepository.count());
    App app = appRepository.findByName("TestApp").orElseThrow();
    
    // 捕获主线程的事务状态
    TransactionStatus mainThreadTxStatus = TransactionAspectSupport.currentTransactionStatus();
    ThreadPoolExecutor executor = new ThreadPoolExecutor(CONCURRENCY, CONCURRENCY, 1, TimeUnit.MINUTES, new SynchronousQueue<>());
    executor.prestartAllCoreThreads();
    
    List<Callable<OptoutMessage>> callables = IntStream.range(0, CONCURRENCY)
            .<Callable<OptoutMessage>>mapToObj(i -> () -> {
                // 将主线程事务上下文绑定到子线程
                TransactionAspectSupport.bindToTransaction(mainThreadTxStatus);
                try {
                    return myService.loggingTransaction(app);
                } finally {
                    // 标记事务需回滚并清理上下文
                    TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
                    TransactionSynchronizationManager.clear();
                }
            })
            .toList();
    
    List<Future<OptoutMessage>> futures = executor.invokeAll(callables);
    executor.shutdownNow();
    
    System.out.println("(after)count: " + logRepository.count());
}

方案三：使用Spring异步框架替代手动线程池

如果业务本身需要异步执行，建议使用Spring的@Async注解配合测试支持，Spring会统一管理异步线程的事务上下文：

1. 在myService.loggingTransaction方法上添加@Async注解
2. 测试类上添加@EnableAsync和@Transactional注解
3. 使用CompletableFuture接收异步方法返回结果，测试结束后Spring会自动回滚所有关联事务

这种方式更贴合Spring生态，避免手动管理线程带来的事务失控问题。

内容的提问来源于stack exchange，提问作者Jiwon