CompletableFuture 与 Spring 事务答案

【问题标题】：CompletableFuture vs Spring TransactionsCompletableFuture 与 Spring 事务
【发布时间】：2016-06-08 07:31:44
【问题描述】：

想法

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

问题

我将列表中的每个项目转换为CompletableFuture 并对它们运行处理任务，然后将它们放回期货数组中。现在使用它的.ofAll 方法（按顺序方法）在所有提交的任务完成后完成future，并返回另一个保存结果的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())
    );
  }

发生了什么？为什么坚持调用没有通过。启动事务的线程是否无法提交事务或者它在哪里丢失？所有处理后的数据都返回正常，一切都很好。我尝试了不同的事务策略，但是如果是这样的话，如何控制哪个线程将完成事务？

有什么建议吗？

【问题讨论】：

sequence 方法呢，能不能展示一下？
在期货列表上执行 allOf 只是胶水代码。
我对此不是 100%，但看起来 Transactional 在方法启动时启动事务并在方法完成时提交，Transactional 不会等待您的回调被触发。你可以用简单的线程来测试。
好吧，我也这么认为。如果我先做List<ProcessData> data = sequence(futures).join() 然后是repository.save(data)，该方法阻塞并提交很好，我不能阻塞，但仍然需要一种方法在事务中从头到尾执行它。
@stjepano 你是对的在执行开始时事务开始事务并在方法结束时提交（或如果运行时错误则回滚）。

标签： java spring multithreading hibernate transactions

【解决方案1】：

如上所述，您的问题的原因是交易结束当达到方法 process(..) 的返回时。

您可以做的是手动创建交易，这可以让您获得完整的控制它何时开始和结束。

删除@Transactional

Autowire TransactionManager 然后在 process(..) 中：

    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;
    }

【讨论】：

重要的是要注意 doWhateverAsync 必须在创建的相同事务上下文中运行。在其他线程中执行操作时应小心，因为相同的上下文可能不可用。
@vishr 只是补充说 CompletableFuture 中发生的任何工作都会有交易，即使我们在未来完成后手动关闭交易，因为根据我的理解，completablefuture 将在一个新线程（或可执行线程池），而不是我们的请求正在处理的线程。并且新线程可能没有相同的上下文。
如何自动完成交易？
这可能不是唯一的原因，似乎事务没有传播到@Transactional-annotated 方法创建的线程：github.com/spring-projects/spring-framework/issues/25439

【解决方案2】：

如果是 Spring Boot Application ，您需要以下配置。

主应用方法应该用@EnableAsync注解。

@Async 注释应该在具有@Transactional 注释的方法的顶部。这对于指示处理将在子线程中进行是必要的。

【讨论】：

在我的情况下，仅添加 @Async 注释无法持久化任何内容。而且我在调试会话中看不到任何插入