运行池外使用 CompletableFuture 和 Spring 交易

Question

我正在尝试使用 CompleteableFutures 将数据库快速加载到内存中。我在方法级别启动 Spring 事务：

@Transactional()
    private void loadErUp() {
        StopWatch sw = StopWatch.createStarted();
        List<CompletableFuture<Void>> calls = new ArrayList<>();
        final ZonedDateTime zdt = ZonedDateTime.now(ZoneId.of(ZoneOffset.UTC.getId())).minusMinutes(REFRESH_OVERLAP);

        for (long i = 1; i < 12 + 1; i++) {
            Long holder = i;
            CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
                this.loadPartition(holder, zdt);
            }, this.forkJoinPool);
            calls.add(future);
        }
        CompletableFuture.allOf(calls.toArray(new CompletableFuture[0])).join();
        log.info("All data refreshed in ({}ms) since:{}", sw.getTime(), zdt.format(DateTimeFormatter.ISO_INSTANT));
    }

然后通过

将每个线程附加到主事务

TransactionSynchronizationManager.setActualTransactionActive(true);

private <T> long loadPartition(long partitionKey, ZonedDateTime zdt) {
        log.debug("Refresh thread start:{}", partitionKey);
        TransactionSynchronizationManager.setActualTransactionActive(true);
        StopWatch sw = StopWatch.createStarted();

        try (Stream<Authority> authorityStream = aSqlRepo.findByPartitionKeyAndLastUpdatedTimeStampAfter(partitionKey, zdt)) {
            long count = authorityStream.peek(a -> {
                this.authorityRepository.set(this.GSON.fromJson(a.getJsonData(), AssetAuthority.class));
            }).count();
            log.info("Partition {} refreshed in ({}ms) with {} items.", partitionKey, sw.getTime(), count);
            return count;
        }
    }

所以我运行这个批处理作业每 30 秒和第 9 个运行我得到 4 个线程然后它挂起 (12*8 运行s = 96) 因为它正在等待游泳池开放。我得到：

Unable to acquire JDBC Connection; Unable to fetch a connection in 30 seconds, none available[size:100; busy:100; idle:0; lastwait:30000].

很明显连接没有提交。我认为这可能是因为我有自己的 ForkJoinPool，但是，我关闭了所有这些线程并且它似乎没有帮助。我还在 loadPartition() 方法下放置了另一个方法，但这似乎也无济于事。还有另一个线程讨论如何让交易工作，但我的工作，他们就是不提交。

Answer 1

如果你想让每个 #loadPartition 运行在它自己的线程和它自己的事务中，你需要：

将 #loadPartition 标记为 @Transactional
调用 proxied #loadPartition 方法以便 @Transactional 起作用。您可以通过自动装配或从另一个代理 class

事务没有传播到异步线程，因为（重要！）.

所以它看起来像：

@Component
public class MyLoaderClass {

    // Autowire in this with constructor injection or @Autowired
    private MyLoaderClass myLoaderClass;

    // Removed @Transactional annotation
    public void loadErUp() {
        myLoaderClass.loadPartition(holder, zdt);
        ...
    }

    // 1) Add the @Transactional annotation to #loadPartition
    // 2) Make public to use self-autowiring (or refactored class, per link above)
    @Transactional
    public <T> long loadPartition(long partitionKey, ZonedDateTime zdt) {
        ...
        // Can remove TransactionSyncManager call
        ...
    }

}

您还需要确保您的批处理作业不会运行无法确保最后一个作业已完成。您可以通过 using the @Scheduled annotation 为您的 table 负载轻松解决此问题，以确保运行不会 "overlap".

运行池外使用 CompletableFuture 和 Spring 交易

Running out of pools using CompleteableFuture and Spring Transaction

java

transactions

spring-transactions

spring-boot

运行 池外使用 CompletableFuture 和 Spring 交易

Running out of pools using CompleteableFuture and Spring Transaction

java

transactions

spring-transactions

spring-boot

运行池外使用 CompletableFuture 和 Spring 交易