使用CompletableFuture执行Stream查询时Spring JPA事务报错求助

核心原因 #

Spring的@Transactional注解依赖动态代理实现事务增强：只有当方法通过Spring代理对象调用时，事务逻辑才会被触发。你在CompletableFuture.supplyAsync的Lambda中直接调用当前类的dbCall方法，属于类内部调用，不会经过代理，导致@Transactional完全失效。而流式查询需要事务保持数据库连接打开，没有事务就会抛出你遇到的错误。

非异步调用时能正常运行，大概率是因为调用getAllData的外层方法本身处于事务中，间接维持了流式查询所需的连接；但异步调用时，任务在新线程执行，脱离了原事务上下文，同时又没触发dbCall的事务注解，因此报错。

可行解决方案 #

方案1：将dbCall拆分到独立Spring Bean（推荐） #

把数据库操作逻辑移到单独的Service Bean中，确保调用时通过代理触发事务：

@Service
public class DataQueryService {
    private final MyRepository myRepository;

    public DataQueryService(MyRepository myRepository) {
        this.myRepository = myRepository;
    }

    @Transactional(readOnly = true)
    public String dbCall(Pair<Long, Long> minMaxPair) {
        try (Stream<MyObject> returnedDataList = myRepository
                .getAllDataInStream(minMaxPair.getFirst(), minMaxPair.getSecond())) {
            
            // 处理返回数据流的业务代码
        }
        return "Data information retrieved for keys between "+minMaxPair.getFirst()+" , "+
                minMaxPair.getSecond();
    }
}

修改原类代码，注入新Bean并调用：

@Service
public class YourService {
    private final DataQueryService dataQueryService;
    private final Logger logger;

    public YourService(DataQueryService dataQueryService, Logger logger) {
        this.dataQueryService = dataQueryService;
        this.logger = logger;
    }

    public void getAllData(List<Pair<Long, Long>> minMaxPairList) throws InterruptedException, ExecutionException {
        Queue<String> globalQueue = new ConcurrentLinkedQueue<>();
        CompletableFuture<?>[] futures = new CompletableFuture<?>[minMaxPairList.size()];
        
        for (int i = 0; i < minMaxPairList.size(); i++) {
            Pair<Long, Long> minMaxPair = minMaxPairList.get(i);
            futures[i] = CompletableFuture
                    .supplyAsync(() -> dataQueryService.dbCall(minMaxPair))
                    .thenAccept(globalQueue::add);
        }
        
        // 等待所有异步任务完成（原代码未做等待，此处补充）
        CompletableFuture.allOf(futures).join();
        
        for (CompletableFuture<?> completableFuture : futures) {
            logger.info("Result: {}", completableFuture.get());
        }
    }
}

方案2：通过AopContext获取代理对象调用方法 #

如果不想拆分Bean，可以开启代理暴露，直接调用代理类的方法触发事务：

1. 在Spring配置类添加注解，开启代理暴露：

@EnableAspectJAutoProxy(exposeProxy = true)
@Configuration
public class AppConfig {
    // 其他配置
}

2. 修改原类中的异步调用逻辑：

futures[i] = CompletableFuture
        .supplyAsync(() -> ((YourService) AopContext.currentProxy()).dbCall(minMaxPair))
        .thenAccept(globalQueue::add);

方案3：手动用TransactionTemplate管理事务 #

直接使用Spring的TransactionTemplate手动控制事务范围，不依赖动态代理：

@Service
public class YourService {
    private final TransactionTemplate transactionTemplate;
    private final MyRepository myRepository;
    private final Logger logger;

    public YourService(TransactionTemplate transactionTemplate, MyRepository myRepository, Logger logger) {
        this.transactionTemplate = transactionTemplate;
        this.myRepository = myRepository;
        this.logger = logger;
    }

    public void getAllData(List<Pair<Long, Long>> minMaxPairList) throws InterruptedException, ExecutionException {
        Queue<String> globalQueue = new ConcurrentLinkedQueue<>();
        CompletableFuture<?>[] futures = new CompletableFuture<?>[minMaxPairList.size()];
        
        for (int i = 0; i < minMaxPairList.size(); i++) {
            Pair<Long, Long> minMaxPair = minMaxPairList.get(i);
            futures[i] = CompletableFuture
                    .supplyAsync(() -> transactionTemplate.execute(status -> {
                        // 事务内执行流式查询
                        try (Stream<MyObject> returnedDataList = myRepository
                                .getAllDataInStream(minMaxPair.getFirst(), minMaxPair.getSecond())) {
                            
                            // 处理返回数据流的业务代码
                        }
                        return "Data information retrieved for keys between "+minMaxPair.getFirst()+" , "+
                                minMaxPair.getSecond();
                    }))
                    .thenAccept(globalQueue::add);
        }
        
        CompletableFuture.allOf(futures).join();
        
        for (CompletableFuture<?> completableFuture : futures) {
            logger.info("Result: {}", completableFuture.get());
        }
    }
}

额外注意点 #

原代码中CompletableFuture.allOf(futures);仅创建了一个聚合Future，并未等待所有异步任务完成，需要调用join()或get()来等待任务结束，否则后续的completableFuture.get()会逐个阻塞等待，逻辑上没问题但不够高效。

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