CompletableFuture:如何将函数应用于多个CompletableFutures？

Question

假设我有3个下载,框架为可完成的期货:

    CompletableFuture<Doc> dl1 = CompletableFuture.supplyAsync(() -> download("file1"));
    CompletableFuture<Doc> dl2 = CompletableFuture.supplyAsync(() -> download("file2"));
    CompletableFuture<Doc> dl3 = CompletableFuture.supplyAsync(() -> download("file3"));

然后所有这些都需要以相同的方式处理

    CompletableFuture<String> s1 = dl1.thenApply(Doc::getFilename);
    CompletableFuture<String> s2 = dl2.thenApply(Doc::getFilename);
    CompletableFuture<String> s3 = dl3.thenApply(Doc::getFilename);

您可以想象要应用的多个功能,所有功能都是并行的.

根据DRY原则,这个例子似乎不合适.所以我正在寻找一种解决方案来定义仅执行3次并行执行的工作流程.

如何实现这一目标？

我尝试过allOf,但这有两个问题:1)它开始阻塞,2)返回类型只能run填充而不是处理它.

Answer 1

Stream.of("file1", "file2", "file3") // or your input in any other format, that can easily be transformed to a stream...
      // .parallel() // well... depends...
      .map(s -> CompletableFuture.supplyAsync(() -> download(s)))
      .map(dl -> dl.thenApply(Doc::getFilename))
      .map(CompletableFuture::join) // if you want to have all the results collected
      .collect(Collectors.toList());

当然也map可以组合两个呼叫.但至少你不写所有的东西x次...如果你不喜欢这个集合,List你也可以在其上调用其他东西,例如.forEach(System.out::println).这样.forEach做的好处是,只要响应可用,就会调用消费者.

或经典之作:只需使用循环和列表/数组作为输入,但您可能需要处理比使用流更多的内容