Spring框架下多线程执行只读DB查询返回null问题排查

我使用Spring+Hibernate（JPA）技术栈、Java 17版本，从PostgreSQL数据库获取Issue流并批量处理：主线程分批收集Issue，通过并行流调用doMeasurements方法执行指标计算，该方法内部需执行数据库查询。单线程执行时一切正常，但切换为并行流后，工作线程的DB查询中find()返回null、count()返回0。

最初发现问题源于DB数据填充与工作线程启动在同一事务，拆分事务后又出现异常：java.lang.IllegalStateException: Illegal pop() with non-matching JdbcValuesSourceProcessingState，怀疑Spring生成的JpaRepository代理非线程安全，想确认该现象是否正常，以及如何实现多线程DB访问。

代码示例如下：

// Issue指的是描述需求的工单，类似Jira或GitHub的Issue
Stream<Issue> issueStream = getIssueStreamFromDB();
Iterator<Issue> issues = issueStream.iterator();

while(issues.hasNext()){
   ParallelBatch<FeatureMinerBean> batch = new ParallelBatch(batchSize);
   while(!batch.isFull && issues.hasNext()){
        Issue issue = issues.next();
        batch.add(new FeatureMinerBean(issue, restOfArguments));
   }
   
    // 并行处理批次内的Issue，执行指标计算
    batch.parallelStream().forEach(b -> doMeasurements(b));

   // 其他代码...
}

1. JpaRepository代理的线程安全性 #

Spring生成的JpaRepository代理本身是线程安全的，问题核心不在代理本身，而在于EntityManager的线程绑定机制：Spring默认通过ThreadLocal将EntityManager绑定到当前线程，并行流的工作线程来自公共线程池，不会继承主线程的EntityManager上下文，导致工作线程无法获取有效持久化上下文，进而出现查询返回空或0的情况。

2. 事务拆分后的异常原因 #

拆分事务后触发的IllegalStateException，是因为主线程的Stream<Issue>（来自数据库查询）与主线程的EntityManager绑定，并行流处理时，工作线程尝试操作同一持久化上下文，引发Hibernate内部状态管理冲突——JdbcValuesSourceProcessingState是Hibernate处理结果集的内部状态，多线程并发操作同一上下文会导致状态栈混乱。

3. 正确的多线程DB访问实现方式 #

方案一：为每个工作线程分配独立事务 #

在doMeasurements方法上添加@Transactional注解，并设置传播行为为REQUIRES_NEW，强制每个调用启动新事务，让Spring为工作线程分配独立的EntityManager：

@Transactional(propagation = Propagation.REQUIRES_NEW)
public void doMeasurements(FeatureMinerBean bean) {
    // 执行数据库查询操作
}

注意：如果该方法是类内部调用，需通过Spring代理对象调用（比如注入自身代理），否则事务注解不会生效。

方案二：提前剥离Issue的持久化上下文 #

主线程分批收集Issue时，将数据从持久化上下文中脱离（调用entityManager.detach(issue)，或通过fetch加载所有关联数据），避免并行流处理时依赖主线程的EntityManager：

// 在获取流或收集Issue时，剥离持久化上下文
Stream<Issue> issueStream = getIssueStreamFromDB().map(issue -> {
    entityManager.detach(issue);
    return issue;
});

方案三：使用自定义上下文感知线程池 #

并行流默认的ForkJoinPool线程无法继承Spring上下文，可自定义线程池并配置上下文感知，提交任务到该线程池执行：

// 配置自定义线程池
@Bean
public TaskExecutor taskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(4);
    executor.setMaxPoolSize(8);
    executor.setThreadNamePrefix("db-worker-");
    executor.initialize();
    return executor;
}

// 使用自定义线程池处理批次
@Autowired
private TaskExecutor taskExecutor;

// 替换并行流代码
List<CompletableFuture<Void>> futures = batch.stream()
    .map(bean -> CompletableFuture.runAsync(() -> doMeasurements(bean), taskExecutor))
    .toList();
CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();

4. 额外注意事项 #

• 禁止在并行流中操作主线程的数据库连接或EntityManager，确保每个线程的DB操作拥有独立资源。
• 控制批次大小和并发线程数，避免数据库压力过载。
• 若开启Hibernate二级缓存，需确保缓存配置线程安全，避免数据不一致。

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