将交易传播到 Forkjoin 提交答案

【问题标题】：Propagate transaction to Forkjoin submit将交易传播到 Forkjoin 提交
【发布时间】：2019-05-25 21:01:12
【问题描述】：

我正在创建一个带有多个线程的 ForkJoinPool 来执行并行流，该流是从 jpa 中的查询执行的，但是我在将事务传播到 ForkJoinPool 的方法提交时遇到问题。

@Transactional(readOnly = true)
public void streamTest() {
    ForkJoinPool customThreadPool = new ForkJoinPool(20);
    try {
    customThreadPool.submit(() ->
         priceRepository.streamAll()
         .parallel()
         .map(p -> this.transform(p))
         .forEach(System.out::println)
         ).get();
    } catch (InterruptedException | ExecutionException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }
}

我收到错误消息：“您正在尝试执行流式查询方法，而没有保持连接打开的周围事务，以便可以实际使用 Stream。确保使用流的代码使用 @Transactional 或声明（只读）事务的任何其他方式。”

如果我取下 ForkJoinPool 来执行流，它就可以正常工作。如何将事务（只读）传播到从 ForkJoinPool 提交的方法的执行，有没有办法？

【问题讨论】：

标签： java spring-boot transactions spring-transactions forkjoinpool

【解决方案1】：

我发现了如何在 ForkJoinPool 的任务中设置事务。我只需要像下面那样使用 TransactionSynchronizationManager。

@Transactional(readOnly = true)
public void streamTest() {
ForkJoinPool customThreadPool = new ForkJoinPool(20);
try {
customThreadPool.submit(() -> {
    TransactionSynchronizationManager.setActualTransactionActive(true);
    TransactionSynchronizationManager.setCurrentTransactionReadOnly(true);
    TransactionSynchronizationManager.initSynchronization();
     priceRepository.streamAll()
     .parallel()
     .map(p -> this.transform(p))
     .forEach(System.out::println);
     }).get();
} catch (InterruptedException | ExecutionException e) {
    // TODO Auto-generated catch block
    e.printStackTrace();
}

}

【讨论】：