Java CompletableFuture thenCompose 与几个异步任务答案

【问题标题】：Java CompletableFuture thenCompose with several async tasksJava CompletableFuture thenCompose 与几个异步任务
【发布时间】：2019-11-14 17:04:17
【问题描述】：

我有一个由 2 个异步步骤组成的过程。第二步基于第一步的结果运行。该过程在循环中启动。挑战在于，第二步是由几个异步任务完成的，它们获取第一步迭代的输出。第一步完成后，我想使用第一步结果启动 n 秒步骤。我使用CompletableFuture 和thenCompose 编写了这段代码。

它有效，但我发现它很复杂，我想知道这是否是正确的方法。我特别想知道二级子任务的管理和使用CompletableFuture.allOf使其像单个CompletableFuture是正确的做法。

public void test() {
    // Gather CompletableFutures to wait for them at the end
    List<CompletableFuture> futures = new ArrayList<>();

    // First steps
    for (int i = 0; i < 10; i++) {
        int finalI = i;
        CompletableFuture<Void> fut = CompletableFuture.supplyAsync(() -> {
            logger.debug("Start step 1 - " + finalI);
            simulateLongProcessing();// just waits for 1 s
            logger.debug("End step 1 - " + finalI);
            return "step1 output - " + finalI;
        }).thenCompose(s -> {
            List<CompletableFuture> subFutures = new ArrayList<>();
            // Second step : Launch several sub-tasks based on the result of the first step
            for (int j = 0; j < 50; j++) {
                final int finalJ = j;
                CompletableFuture<String> f = CompletableFuture.supplyAsync(() -> {
                    logger.debug("Start - step 2 : " + s + " | " + finalJ);
                    simulateLongProcessing();
                    logger.debug("End - step 2 : " + s + " | " + finalJ);
                    return "step2 output - " + s + " | " + finalJ;
                });
                subFutures.add(f);
            }
            return CompletableFuture.allOf(subFutures.toArray(new CompletableFuture[0]));
        });
        futures.add(fut);
    }

    // Wait for the completion
    for (CompletableFuture future : futures) {
        future.join();
    }
}

【问题讨论】：

我认为让这看起来复杂的原因是循环创建了许多未来。如果重构方法，使嵌套循环或thenCompose lambda 转到单独的方法，它看起来会更简单。

标签： java concurrency completable-future

【解决方案1】：

不要在传递给thenCompose 的函数中执行CompletableFuture.supplyAsync，因为您可以将相同的依赖函数与直接的thenApplyAsync 链接起来。

通过thenApplyAsync 链接依赖函数允许您在第一步完成之前获取代表这些步骤的CompletableFuture 实例，因此您可以将它们全部收集到您的List 中以等待它们在最后完成并不要不需要通过CompletableFuture.allOf 创建复合期货。

public void test() {
    // Gather CompletableFutures to wait for them at the end
    List<CompletableFuture<?>> futures = new ArrayList<>();

    for (int i = 0; i < 10; i++) {
        int finalI = i;
        CompletableFuture<String> step1 = CompletableFuture.supplyAsync(() -> {
            logger.debug("Start step 1 - " + finalI);
            simulateLongProcessing();// just waits for 1 s
            logger.debug("End step 1 - " + finalI);
            return "step1 output - " + finalI;
        });
        // Second step : Chain several sub-tasks based on the result of the first step
        for (int j = 0; j < 50; j++) {
            final int finalJ = j;
            futures.add(step1.thenApplyAsync(s -> {
                logger.debug("Start - step 2 : " + s + " | " + finalJ);
                simulateLongProcessing();
                logger.debug("End - step 2 : " + s + " | " + finalJ);
                return "step2 output - " + s + " | " + finalJ;
            }));
        }
    }

    // Wait for the completion
    for (CompletableFuture<?> future : futures) {
        future.join();
    }
}

【讨论】：