在带有两个未来的循环中使用CompletableFuture来合并每个循环迭代

Question

我有如下代码：

testMethod(List<String> ids)  {
    List<CompletableFuture<ResultThree>> resultThreeList = new ArrayList<>();
    
    for(String id : ids) {
        CompletableFuture<ResultOne> resultOne = AynchOne(id);
        CompletableFuture<ResultTwo> resultTwo = AynchTwo(id);
        
    CompletableFuture<ResultThree> resultThree =  resultOne.thenCombine(resultTwo, (ResultOne a, ResultTwo b) -> computeCombinedResultThree(a, b)); 
    
    resultThreeList.add(resultThree);
    }
    // PROCESS RESULTS HERE
}

class ResultOne {
    boolean goodResult;
    String id;

    ResultOne(String promId) {
        this.goodResult = true;
        this.id = promId;
    }
}

class ResultTwo {
    boolean goodResult;
    String id;

    ResultTwo(String promId) {
        this.goodResult = true;
        this.id = promId;
    }

class ResultThree() {
        boolean goodResult;
        String = id;
    }

private ResultThree computeCombinedResultThree(ResultOne r1,  ResultTwo r2) { 
   ResultThree resultThree = new ResultThree();
    resultThree.id = r1.id;
    resultThree.goodResult = r1.goodResult && r2.goodResult;

    return resultThree;
}

，我需要能够将结果resultOne和resultTwo放在一起，这样对于每次迭代，在完成整个同步执行时，我有一个（我猜）数组或映射，我可以随后处理，其中数组中的一个对象有相应的id，该id有一个true或false（表示两个布尔值与单独对象的和）。

根据读者的反馈，我已经完成了代码，可以合并两个原始期货，并组合每次迭代的所有结果以获得整个期货循环。此时我只需要处理结果。

我想也许我需要另一个完整的未来？这个可能是这样的（放在上面，我这里有“//过程结果”）：

CompletableFuture<Void> future = resultThreeList
  .thenRun(() -> forwardSuccesses(resultThreeList));

future.get();

ForwardSuccessfulls（）将遍历resultThreeList，将成功的ID转发给另一个进程，但不是这样做的。谢谢你的建议。谢谢

Answer 1

在for循环中，你会立即得到CompletableFutures作为回报。在后台，一些魔法发生了，你想等到两者都完成。

因此，在两个CompletableFuture都返回后，通过调用CompletableFuture引起阻塞等待。获取一个长值和一个时间单位。如果只调用get而不调用任何参数，您将永远等待。

明智地选择超时和JDK。JDK 8可能不提供get with timeout。此外，JDK 8不再受支持。JDK 11现在是长期支持，最近的编译器不再将JDK 8作为目标。

我真的敦促你们阅读关于CompletableFuture的肮脏细节，以及它与Future的区别，尤其是关于线程控制（比如取消）的细节。由于不知道CompletableFuture的底层提供者，我还假设查询一个ID是对资源的浪费，吞吐量非常有限。但这是另一个问题。

Answer 2

在我看来，你不需要三个不同的resulttoneResultTwoResultTwo三个类，因为它们定义了相同的类型，所以我将为Result替换它们。

假设您只想转发成功，我在Result类中添加了一个简短的isGoodResult（）方法，用作流的谓词：

class Result {
    public boolean goodResult;
    public String id;
// ...
    public boolean isGoodResult() {
        return this.goodResult;
    }
}

我还建议去掉循环，将其替换为流，以使代码更流畅。

forwardSuccess应该严格接受列表吗

void testMethod(List<String> ids)  {
    final List<Result> results = ids.stream()
        .parallel()
        .map(id -> asynchOne(id).thenCombine(
            asynchTwo(id), 
            (r1, r2) -> computeCombinedResult(r1, r2)))
        .map(CompletableFuture::join)
        .filter(Result::isGoodResult)
        .collect(Collectors.toList());

    // PROCESS RESULTS HERE
    forwardSuccesses(results);
}

应该转发成功偷懒，接受CompletableFuture

void testMethod(List<String> ids)  {
    final List<CompletableFuture<Result>> futures = ids.stream()
        .parallel()
        .map(id -> asynchOne(id).thenCombine(
            asynchTwo(id), 
            (r1, r2) -> computeCombinedResult(r1, r2)))
        .collect(Collectors.toList());

    final CompletableFuture<List<Result>> asyncResults =
    CompletableFuture.allOf(futures.stream().toArray(CompletableFuture[]::new))
            .thenApply(__ -> futures 
                    .stream()
                    .map(CompletableFuture::join)
                    .filter(Result::isGoodResult)
                    .collect(Collectors.toList()));

    // PROCESS RESULTS HERE
    forwardSuccessesAsync(asyncResults);
}

Answer 3

这就是你到现在为止所取得的成绩：

List<CompletableFuture<ResultThree>> resultThreeList = new ArrayList<>(ids.size());
for (String id : ids) {
    CompletableFuture<ResultOne> resultOne = aynchOne(id);
    CompletableFuture<ResultTwo> resultTwo = aynchTwo(id);

    CompletableFuture<ResultThree> resultThree = resultOne.thenCombine(resultTwo, this::computeCombinedResultThree);
    resultThreeList.add(resultThree);
}

现在你需要做的就是转换这个列表

CompletableFuture<List<ResultThree>> combinedCompletables =
        CompletableFuture.allOf(resultThreeList.toArray(new CompletableFuture<?>[0]))
                .thenApply(v -> resultThreeList.stream()
                        .map(CompletableFuture::join)
                        .collect(Collectors.toList())
                );

或者用类似的东西

CompletableFuture<List<ResultThree>> combinedCompletables =
        CompletableFuture.supplyAsync(() -> resultThreeList.stream().map(this::safeGet).collect(Collectors.toList()));

其中，safeGet是一个只调用future的方法。get（）并捕获可能发生的异常——由于这些异常，您不能在lambda中调用get（）。

现在，您可以使用然后accept（）处理此列表：

try {
    combinedCompletables.thenAccept(this::forwardSuccesses).get(30, TimeUnit.SECONDS);
} catch (InterruptedException | ExecutionException | TimeoutException e) {
    e.printStackTrace();
}

同样，捕捉到的异常是由于调用了get（）。

附带说明，我真的不明白为什么有三个结果类，因为您所需要的——至少对于这部分代码——是id和结果状态。我会为此引入一个接口（Result？），并且只处理它。