深入理解CompletableFuture：实现任务的顺序执行与结果收集-java教程-PHP中文网

深入理解completablefuture：实现任务的顺序执行与结果收集

本文旨在探讨如何使用Java的CompletableFuture实现一系列异步任务的顺序执行，并将所有任务的结果收集到一个列表中。我们将分析常见的陷阱，如不当的线程管理和并发执行问题，并提供两种优雅且高效的解决方案，确保任务按预期顺序完成并正确汇总结果。

1. 问题背景与挑战

在异步编程中，CompletableFuture是处理并发任务的强大工具。然而，当面临需要严格顺序执行的异步任务链，并且需要收集每个任务的结果时，可能会遇到一些挑战。例如，业务场景可能要求前一个任务完成后，后一个任务才能开始，同时我们希望将所有任务的计算结果汇总到一个集合中。

考虑一个耗时的业务处理函数，它返回一个CompletionStage<Integer>：

import java.time.LocalDateTime;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;

public class SequentialTaskProcessor {

    private CompletionStage<Integer> process(int a) {
        return CompletableFuture.supplyAsync(() -> {
            System.err.printf("%s dispatch %d\n", LocalDateTime.now(), a);
            // 模拟长时间运行的业务处理
            try {
                Thread.sleep(10); // 增加延迟以观察效果
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            return a + 10;
        }).whenCompleteAsync((e, t) -> {
            if (t != null)
                System.err.printf("!!! error processing '%d' !!!\n", a);
            System.err.printf("%s finish %d\n", LocalDateTime.now(), e);
        });
    }

登录后复制

我们的目标是多次调用process函数，确保它们按顺序执行，并将每次的结果收集到一个List<Integer>中。

1.1 常见误区：thenApplyAsync与内部join()

一种直观的尝试是使用thenApplyAsync并在其内部调用process(element).toCompletableFuture().join()。

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

// ... (process方法同上)

public void firstApproach() {
    List<Integer> arr = IntStream.range(1, 10).boxed().collect(Collectors.toList());
    CompletionStage<List<Integer>> result = CompletableFuture.completedFuture(new ArrayList<>());

    for (Integer element : arr) {
        result = result.thenApplyAsync((ret) -> {
            // 在thenApplyAsync内部阻塞等待前一个CompletableFuture完成
            Integer a = process(element).toCompletableFuture().join(); 
            ret.add(a);
            return ret;
        });
    }

    List<Integer> computeResult = result.toCompletableFuture().join();
    System.out.println("First approach results: " + computeResult);
}

登录后复制

问题分析： 虽然这种方法能够实现顺序执行并收集结果，但它效率低下。thenApplyAsync本身会在一个线程池中执行其回调，而回调内部的process(element).toCompletableFuture().join()又会阻塞这个线程，直到process方法返回的CompletableFuture完成。这意味着一个逻辑步骤可能间接占用两个线程资源（一个用于thenApplyAsync的回调，另一个用于process内部的异步任务），造成线程资源的浪费和不必要的阻塞。观察输出日志，会发现dispatch和finish的时间戳是严格顺序的，但线程利用率不高。

1.2 常见误区：thenCombineAsync的并发陷阱

另一种尝试是使用thenCombineAsync，期望它能将前一个阶段的结果与新任务的结果结合：

// ... (process方法同上)

public void secondApproach() {
    List<Integer> arr = IntStream.range(1, 10).boxed().collect(Collectors.toList());
    CompletionStage<List<Integer>> result = CompletableFuture.completedFuture(new ArrayList<>());

    for (Integer element : arr) {
        // process(element) 在这里被立即调用，而非等待前一个阶段完成
        result = result.thenCombineAsync(process(element), (array, ret) -> { 
            array.add(ret); 
            return array; 
        });
    }

    List<Integer> computeResult = result.toCompletableFuture().join();
    System.out.println("Second approach results: " + computeResult);
}

登录后复制

问题分析： 这种方法会导致任务并发执行，而非顺序执行。thenCombineAsync的第二个参数CompletionStage<U> other在方法调用时就会被评估并启动。这意味着在循环中，所有的process(element)调用几乎是同时发起的，它们会并发执行。观察输出日志，会发现dispatch的时间戳是交错的，这违反了顺序执行的要求。thenCombineAsync适用于两个独立的异步任务都完成后再进行合并的场景，而不是链式顺序执行的场景。

2. 解决方案：顺序链式执行与结果收集

为了实现任务的顺序执行并高效地收集结果，我们需要利用CompletableFuture提供的更高级的组合方法，特别是thenCompose。

2.1 方案一：使用外部列表收集结果

这种方法通过thenCompose确保任务顺序执行，并使用thenAccept将结果添加到循环外部维护的列表中。

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

// ... (process方法同上)

public class SequentialTaskProcessor {
    // ... process 方法 ...

    public void solutionOne() {
        List<Integer> arr = IntStream.range(1, 10).boxed().collect(Collectors.toList());

        // 初始化一个表示链式操作开始的CompletableFuture，其结果类型为Void
        CompletionStage<Void> loopStage = CompletableFuture.completedFuture(null);
        final List<Integer> resultList = new ArrayList<>(); // 外部结果列表

        for (Integer element : arr) {
            loopStage = loopStage
                    // thenCompose确保前一个阶段完成后，才执行process(element)
                    .thenCompose(v -> process(element)) 
                    // thenAccept将process的结果添加到外部列表中，并返回CompletionStage<Void>
                    .thenAccept(resultList::add); 
        }

        // 阻塞等待所有任务完成
        loopStage.toCompletableFuture().join(); 

        System.out.println("Solution One results: " + resultList);
    }

    public static void main(String[] args) {
        SequentialTaskProcessor processor = new SequentialTaskProcessor();
        System.out.println("--- Running Solution One ---");
        processor.solutionOne();
        System.out.println("\n--- Running Solution Two ---");
        processor.solutionTwo();
    }
}

登录后复制

原理详解：

影像之匠PixPretty

商业级AI人像后期软件，专注于人像精修，色彩调节及批量图片编辑，支持Windows、Mac多平台使用。适用于写真、婚纱、旅拍、外景等批量修图场景。

299

查看详情

CompletionStage<Void> loopStage = CompletableFuture.completedFuture(null);：我们从一个已完成的CompletableFuture开始，其结果类型为Void。这提供了一个初始的“钩子”来启动任务链。
loopStage = loopStage.thenCompose(v -> process(element))：thenCompose是这里的关键。它接收一个函数，该函数返回一个新的CompletionStage。这意味着process(element)只会在loopStage（即前一个任务）完成后才会被调用并开始执行。这确保了任务的严格顺序性。thenCompose的作用是将CompletionStage<Void>（来自loopStage）和CompletionStage<Integer>（来自process）的结果扁平化为一个新的CompletionStage<Integer>。
.thenAccept(resultList::add)：在process(element)完成并产生结果后，thenAccept会异步地将该结果添加到resultList中。thenAccept本身返回一个CompletionStage<Void>，这使得loopStage可以继续作为链的下一个开始点，而不必传递一个累积的列表。
loopStage.toCompletableFuture().join()：最后，我们阻塞等待整个任务链的最终阶段完成。此时，resultList将包含所有任务的顺序结果。

这种方法简洁且高效，避免了不必要的阻塞和线程浪费。

2.2 方案二：在链中传递并累积列表

另一种方法是在CompletableFuture链中直接传递并累积结果列表。

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

// ... (process方法同上)

public class SequentialTaskProcessor {
    // ... process 方法 ...

    public void solutionTwo() {
        List<Integer> arr = IntStream.range(1, 10).boxed().collect(Collectors.toList());
        // 初始化一个携带空列表的CompletableFuture
        CompletionStage<List<Integer>> listStage = CompletableFuture.completedFuture(new ArrayList<>());

        for (Integer element : arr) {
            listStage = listStage
                    // thenCompose确保前一个阶段完成后，才执行process(element)
                    .thenCompose(list -> process(element) 
                            // thenAccept将process的结果添加到当前列表
                            .thenAccept(list::add) 
                            // thenApply将CompletionStage<Void>转换回CompletionStage<List<Integer>>
                            .thenApply(v -> list) 
                    );
        }

        // 阻塞等待所有任务完成，并获取最终的列表
        List<Integer> resultList = listStage.toCompletableFuture().join(); 

        System.out.println("Solution Two results: " + resultList);
    }
    // ... main 方法 ...
}

登录后复制

原理详解：

CompletionStage<List<Integer>> listStage = CompletableFuture.completedFuture(new ArrayList<>());：我们从一个包含空列表的CompletableFuture开始，这个列表将作为结果的累积器。
listStage = listStage.thenCompose(list -> ...)：同样使用thenCompose来确保顺序执行。这里的list参数是前一个阶段传递过来的结果列表。
process(element).thenAccept(list::add)：在thenCompose的函数内部，我们启动process(element)任务。当它完成时，使用thenAccept将结果添加到当前list中。
.thenApply(v -> list)：这是关键一步。thenAccept返回的是CompletionStage<Void>，但为了将list传递给下一个迭代，我们需要将其结果类型转换回CompletionStage<List<Integer>>。thenApply(v -> list)实现了这一点：它在thenAccept完成后被调用，并简单地返回当前的list对象，从而将列表传递给链中的下一个thenCompose。
List<Integer> resultList = listStage.toCompletableFuture().join();：最终，整个链完成时，listStage的结果就是包含了所有累积结果的列表。

3. 总结与注意事项

两种解决方案都能够有效地实现异步任务的顺序执行和结果收集，并且都避免了线程阻塞和并发执行的问题。

方案一（外部列表）：
- 优点：代码逻辑相对直观，loopStage只关心任务的完成状态（Void），结果列表在外部维护。
- 适用场景：当任务链的中间结果不需要在CompletableFuture链中传递，只需最终汇总时。
方案二（链中传递列表）：
- 优点：结果列表直接作为CompletableFuture链的一部分进行传递和累积，整个操作封装在一个CompletableFuture中，最终结果直接从CompletableFuture获取。
- 适用场景：当需要将累积的结果作为链中下一个任务的输入，或者更倾向于将所有状态变化封装在CompletableFuture链内部时。

注意事项：

异常处理：在实际应用中，需要为CompletableFuture链添加适当的异常处理机制，例如使用exceptionally、handle等方法来处理任务执行过程中可能出现的错误。
线程池管理：CompletableFuture默认使用ForkJoinPool.commonPool()。对于长时间运行或IO密集型任务，建议为supplyAsync、thenApplyAsync等方法指定自定义的Executor，以更好地控制线程资源，避免阻塞公共线程池。
任务原子性：确保process方法内部的业务逻辑是线程安全的，如果它操作共享资源，需要额外的同步机制。本文的重点在于CompletableFuture的链式调用，而非process方法本身的线程安全性。

通过理解thenCompose的扁平化特性和thenAccept/thenApply的组合使用，我们可以更灵活、高效地构建复杂的异步任务流，满足各种顺序执行和结果收集的需求。

以上就是深入理解CompletableFuture：实现任务的顺序执行与结果收集的详细内容，更多请关注php中文网其它相关文章！