resolve/handle List<CompletableFuture<List<Object>>> in Java 8 的最佳方式

Question

代码： Spring bean 片段：

@Component
class myService {

 private CSVFileProcessor csvFileProcessor;

 public myService(CSVFileProcessor csvFileProcessor) {

         this.csvFileProcessor=csvFileProcessor;

    }

/* input: List of promises/futures obtained by reading 
 *   csv files from S3 bucket
 * 
 * output: List of promises of list of parsed csv 
 *   documents converted to POJO's of type Document
 *  
 *
 */
  @Async
  public List<CompletableFuture<List<Document>>> createDocumentObjects(
      List<CompletableFuture<ResponseBytes<GetObjectResponse>>> documentsFuture) {

 return documentsFuture.stream().map(myDocument-> myDocument.thenCompose(document-> CompletableFuture.supplyAsync(
                 () ->   csvFileProcessor.parseObjects(document)))).collect(Collectors.toList());

}

所以在服务层我注入上面的 bean 并尝试做类似的事情：

List<CompletableFuture<List<Document>>> listOfPromises= injectedBean.createDocumentObjects(input);

//The below code throws NPE(Null pointer exception) 
List<List<Document>> myDocuments =listOfPromises.stream().map(CompletableFuture::join).collect(Collectors.toList());

listOfPromises 似乎为空。我检查了 csvFileProcessor class 的日志，输入正在按预期进行处理。尝试向未来的调用添加异常和处理块，在尝试解决承诺时，除了同一个 NPE 之外的那些地方仍然没有抛出异常。

我确信我遗漏了一些微不足道的东西，任何正确方向的指导都会非常有帮助。

Answer 1

根据 @Async documentation:

In terms of target method signatures, any parameter types are supported. However, the return type is constrained to either void or Future. In the latter case, you may declare the more specific ListenableFuture or CompletableFuture types which allow for richer interaction with the asynchronous task and for immediate composition with further processing steps.

你的方法 returns 两者都不是，所以 Spring 不知道如何处理它。不幸的是，它没有给出错误，而是接受了调用，将方法执行推迟到一个单独的线程并立即 returns null.

但是你似乎并不需要 @Async 这里，因为你的方法正在处理异步执行本身（通过组合和 supplyAsync()），所以你可以只删除注释解决问题。

resolve/handle List<CompletableFuture<List<Object>>> in Java 8 的最佳方式

Optimal way to resolve/handle List<CompletableFuture<List<Object>>> in Java 8

java

java-8

java-stream

completable-future