CompletableFuture vs Spring Transactions

Question

理念

我有一个处理方法,它接收一个项目列表并使用外部Web服务异步处理它们.处理步骤还会在处理时保留数据.在整个过程结束时,我希望将整个过程与每个处理结果一起保持.

问题

我将列表中的每个项目转换为CompletableFuture并对它们运行处理任务,并将它们放回到未来的数组中.现在使用其.ofAll方法(按顺序方法)完成所有提交的任务完成后的未来,并返回CompletableFuture保存结果的另一个任务.

当我想得到那个结果时,我会调用.whenComplete(..),并且想要将返回的结果作为数据设置到我的实体中,然后保存到数据库,但是存储库保存调用什么都不做,继续线程只是继续运行,它不会去通过存储库保存调用.

 @Transactional
 public void process(List<Item> items) {
   List<Item> savedItems = itemRepository.save(items);

   final Process process = createNewProcess();

   final List<CompletableFuture<ProcessData>> futures = savedItems.stream()
     .map(item -> CompletableFuture.supplyAsync(() -> doProcess(item, process), executor))
     .collect(Collectors.toList());

   sequence(futures).whenComplete((data, throwable) -> {
     process.setData(data);
     processRepository.save(process); // <-- transaction lost?
     log.debug("Process DONE"); // <-- never reached
   });
  }

序列方法

 private static <T> CompletableFuture<List<T>> sequence(List<CompletableFuture<T>> futures) {
    CompletableFuture<Void> allDoneFuture =
      CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()]));
    return allDoneFuture.thenApply(v ->
      futures.stream().map(CompletableFuture::join).collect(Collectors.toList())
    );
  }

怎么了？为什么持续的呼叫没有通过.启动事务的线程是否无法提交事务或丢失的位置？所有处理过的数据都很好,并且都很好.我尝试过不同的交易策略,但如果是这样的话,如何控制哪个线程完成交易呢？

有什么建议？

Answer 1

如上所述,您遇到问题的原因是,当达到方法进程(..)的返回时,事务结束.

你可以做的是手动创建交易,让你完全控制它的开始和结束时间.

删除@Transactional

然后在进程(..)中自动装配TransactionManager:

    TransactionDefinition txDef = new DefaultTransactionDefinition();
    TransactionStatus txStatus = transactionManager.getTransaction(txDef);
    try {
    //do your stuff here like
        doWhateverAsync().then(transactionManager.commit(txStatus);)
    } catch (Exception e) {
        transactionManager.rollback(txStatus);
        throw e;
    }