CompletableFuture 的链式调用与任务组合

一、前言

在上一篇中，我们了解到 CompletableFuture 相比传统 Future 的核心优势之一，就是强大的任务组合能力。它通过丰富的 API 支持任务的串行、并行、依赖组合，让复杂的异步流程编排变得简洁优雅。本篇作为核心 API 篇，将深入拆解 CompletableFuture 的链式调用与任务组合 API，结合实战场景讲解其使用方法与核心原理。

二、基础链式调用API

串行执行是异步编程中最基础的场景：任务 A 执行完成后，自动触发任务 B 的执行，且任务 B 可以复用任务 A 的结果。CompletableFuture 提供了一组以 then 开头的 API 来实现串行任务链，根据任务是否需要参数、是否有返回值，可分为三类核心方法。

1、无参数无返回值：thenRun ()/thenRunAsync ()

thenRun() 方法用于前序任务完成后，执行一个无参数、无返回值的后续任务 ，适用于 “任务执行完毕后做一些收尾操作” 的场景（如日志记录、资源释放）。

它有两个重载方法，核心区别在于执行后续任务的线程不同 ：

// 使用前序任务的线程执行后续任务
public CompletableFuture<Void> thenRun(Runnable action)
// 使用默认线程池（ForkJoinPool.commonPool()）或自定义线程池执行后续任务
public CompletableFuture<Void> thenRunAsync(Runnable action)
public CompletableFuture<Void> thenRunAsync(Runnable action, Executor executor)

实战示例：异步查询订单后记录日志

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class ThenRunDemo {
    private static final ExecutorService executor = Executors.newFixedThreadPool(2);
    public static void main(String[] args) {
        // 1. 异步查询订单（前序任务，有返回值）
        CompletableFuture<String> orderFuture = CompletableFuture.supplyAsync(() -> {
            System.out.println("前序任务线程：" + Thread.currentThread().getName());
            try {
                Thread.sleep(500);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return "订单号：ORDER_12345";
        }, executor);
        // 2. 前序任务完成后，执行日志记录（无参数无返回值）
        CompletableFuture<Void> logFuture = orderFuture.thenRun(() -> {
            System.out.println("thenRun 任务线程：" + Thread.currentThread().getName());
            System.out.println("订单查询完成，记录操作日志");
        });
        // 等待任务完成
        logFuture.join();
        executor.shutdown();
    }
}

运行结果：

前序任务线程：pool-1-thread-1
thenRun 任务线程：pool-1-thread-1
订单查询完成，记录操作日志

结论： thenRun() 未使用 Async 后缀时，后续任务与前序任务共用同一个线程；若使用 thenRunAsync() ，则后续任务会提交到线程池执行，二者运行在不同线程。

2、接收参数无返回值：thenAccept ()/thenAcceptAsync ()

thenAccept() 方法用于前序任务完成后，接收前序任务的结果作为参数，执行一个无返回值的后续任务，适用于 “消费前序任务结果” 的场景（如打印结果、数据落盘）。

核心方法定义：

// 消费前序任务结果，无返回值
public CompletableFuture<Void> thenAccept(Consumer<? super T> action)
// 异步消费前序任务结果
public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action)
public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action, Executor executor)

实战示例：异步查询用户信息后打印结果

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class ThenAcceptDemo {
    private static final ExecutorService executor = Executors.newFixedThreadPool(2);
    // 模拟用户信息
    static class User {
        private String userId;
        private String userName;
        public User(String userId, String userName) {
            this.userId = userId;
            this.userName = userName;
        }
        @Override
        public String toString() {
            return "User{" + "userId='" + userId + '\\'' + ", userName='" + userName + '\\'' + '}';
        }
    }
    public static void main(String[] args) {
        // 1. 异步查询用户信息
        CompletableFuture<User> userFuture = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(500);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return new User("1001", "张三");
        }, executor);
        // 2. 接收用户信息并打印（消费结果，无返回值）
        userFuture.thenAccept(user -> {
            System.out.println("获取到用户信息：" + user);
        }).join();
        executor.shutdown();
    }
}

运行结果：

获取到用户信息：User{userId='1001', userName='张三'}

3、接收参数有返回值：thenApply ()/thenApplyAsync ()

thenApply() 是串行调用中最常用的 API，它支持接收前序任务的结果作为参数，执行有返回值的后续任务，适用于 “数据转换、结果加工” 的场景（如从用户信息中提取用户名、对订单金额计算优惠）。

核心方法定义：

// 接收前序结果，转换为新的结果返回
public <U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn)
// 异步执行转换任务
public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn)
public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn, Executor executor)

实战示例：用户信息查询→提取用户名→生成欢迎语

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class ThenApplyDemo {
    private static final ExecutorService executor = Executors.newFixedThreadPool(3);
    static class User {
        private String userId;
        private String userName;
        public User(String userId, String userName) {
            this.userId = userId;
            this.userName = userName;
        }
        public String getUserName() {
            return userName;
        }
    }
    public static void main(String[] args) {
        // 任务1：查询用户信息
        CompletableFuture<User> userFuture = CompletableFuture.supplyAsync(() -> {
            System.out.println("任务1线程：" + Thread.currentThread().getName());
            try {
                Thread.sleep(500);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return new User("1001", "张三");
        }, executor);
        // 任务2：提取用户名（接收User对象，返回String）
        CompletableFuture<String> nameFuture = userFuture.thenApplyAsync(user -> {
            System.out.println("任务2线程：" + Thread.currentThread().getName());
            return user.getUserName();
        }, executor);
        // 任务3：生成欢迎语（接收String，返回String）
        CompletableFuture<String> welcomeFuture = nameFuture.thenApplyAsync(name -> {
            System.out.println("任务3线程：" + Thread.currentThread().getName());
            return "欢迎你，" + name + "！";
        }, executor);
        // 获取最终结果
        System.out.println(welcomeFuture.join());
        executor.shutdown();
    }
}

运行结果：

任务1线程：pool-1-thread-1
任务2线程：pool-1-thread-2
任务3线程：pool-1-thread-3
欢迎你，张三！

4、关键区别：带 Async 与不带 Async 的差异

通过上述示例可以总结出 Async 后缀的核心作用：决定后续任务的执行线程 。

方法类型

执行线程

适用场景

注意 ：使用 Async 方法时，若指定了自定义线程池，则优先使用自定义线程池；否则使用 ForkJoinPool.commonPool() 。

三、多任务并行组合 API

在实际开发中，我们经常需要并行执行多个无依赖的异步任务，再对结果进行聚合。比如电商首页需要并行查询用户信息、待支付订单、会员积分，然后统一返回。CompletableFuture 提供了 allOf() 和 anyOf() 两个静态方法，完美解决多任务并行组合问题。

1、allOf ()：等待所有任务完成

allOf() 方法接收一个或多个 CompletableFuture 对象作为参数，返回一个 CompletableFuture 。它的核心特性是：等待所有传入的任务都执行完成后，当前任务才会完成，但不返回任何结果，需要手动获取每个任务的结果。

核心方法定义：

public static CompletableFuture<Void> allOf(CompletableFuture<?>… cfs)

实战示例：电商首页多源数据并行查询与聚合

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class AllOfDemo {
    private static final ExecutorService executor = Executors.newFixedThreadPool(3);
    // 模拟查询用户信息
    private static CompletableFuture<String> queryUserInfo(String userId) {
        return CompletableFuture.supplyAsync(() -> {
            System.out.println("查询用户信息：" + Thread.currentThread().getName());
            try {
                Thread.sleep(800);
            } catch (InterruptedException e) {
                throw new RuntimeException("查询用户信息失败", e);
            }
            return "用户信息：userId=" + userId + ", 姓名=张三";
        }, executor);
    }
    // 模拟查询待支付订单
    private static CompletableFuture<String> queryPendingOrders(String userId) {
        return CompletableFuture.supplyAsync(() -> {
            System.out.println("查询待支付订单：" + Thread.currentThread().getName());
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                throw new RuntimeException("查询订单失败", e);
            }
            return "待支付订单：userId=" + userId + ", 数量=2";
        }, executor);
    }
    // 模拟查询会员积分
    private static CompletableFuture<String> queryUserPoints(String userId) {
        return CompletableFuture.supplyAsync(() -> {
            System.out.println("查询会员积分：" + Thread.currentThread().getName());
            try {
                Thread.sleep(600);
            } catch (InterruptedException e) {
                throw new RuntimeException("查询积分失败", e);
            }
            return "会员积分：userId=" + userId + ", 积分=1000";
        }, executor);
    }
    public static void main(String[] args) {
        String userId = "1001";
        long start = System.currentTimeMillis();
        // 1. 并行提交三个任务
        CompletableFuture<String> userFuture = queryUserInfo(userId);
        CompletableFuture<String> orderFuture = queryPendingOrders(userId);
        CompletableFuture<String> pointsFuture = queryUserPoints(userId);
        // 2. 等待所有任务完成
        CompletableFuture<Void> allFuture = CompletableFuture.allOf(userFuture, orderFuture, pointsFuture);
        // 3. 所有任务完成后，聚合结果
        CompletableFuture<List<String>> resultFuture = allFuture.thenApplyAsync(v -> {
            List<String> resultList = new ArrayList<>();
            resultList.add(userFuture.join());
            resultList.add(orderFuture.join());
            resultList.add(pointsFuture.join());
            return resultList;
        }, executor);
        // 4. 获取最终聚合结果
        List<String> result = resultFuture.join();
        System.out.println("首页数据聚合结果：");
        result.forEach(System.out::println);
        System.out.println("总耗时：" + (System.currentTimeMillis() – start) + "ms");
        executor.shutdown();
    }
}

运行结果：

查询用户信息：pool-1-thread-1
查询待支付订单：pool-1-thread-2
查询会员积分：pool-1-thread-3
首页数据聚合结果：
用户信息：userId=1001, 姓名=张三
待支付订单：userId=1001, 数量=2
会员积分：userId=1001, 积分=1000
总耗时：1015ms

核心结论：

2、anyOf ()：任意一个任务完成即返回

anyOf() 方法同样接收多个 CompletableFuture 对象作为参数，但它的核心特性是：只要有任意一个子任务完成，当前任务就会完成，并返回该子任务的结果。适用于 “多渠道调用，取最快响应结果” 的场景（如调用多个支付接口，选择最快返回的一个）。

核心方法定义：

public static CompletableFuture<Object> anyOf(CompletableFuture<?>… cfs)

实战示例：多支付渠道调用，取最快响应结果

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class AnyOfDemo {
    private static final ExecutorService executor = Executors.newFixedThreadPool(3);
    // 模拟支付渠道A
    private static CompletableFuture<String> payByChannelA(String orderId) {
        return CompletableFuture.supplyAsync(() -> {
            System.out.println("渠道A处理支付：" + Thread.currentThread().getName());
            try {
                Thread.sleep(1000); // 模拟耗时1秒
            } catch (InterruptedException e) {
                throw new RuntimeException("渠道A支付失败", e);
            }
            return "渠道A支付成功，订单号：" + orderId;
        }, executor);
    }
    // 模拟支付渠道B
    private static CompletableFuture<String> payByChannelB(String orderId) {
        return CompletableFuture.supplyAsync(() -> {
            System.out.println("渠道B处理支付：" + Thread.currentThread().getName());
            try {
                Thread.sleep(500); // 模拟耗时0.5秒
            } catch (InterruptedException e) {
                throw new RuntimeException("渠道B支付失败", e);
            }
            return "渠道B支付成功，订单号：" + orderId;
        }, executor);
    }
    // 模拟支付渠道C
    private static CompletableFuture<String> payByChannelC(String orderId) {
        return CompletableFuture.supplyAsync(() -> {
            System.out.println("渠道C处理支付：" + Thread.currentThread().getName());
            try {
                Thread.sleep(800); // 模拟耗时0.8秒
            } catch (InterruptedException e) {
                throw new RuntimeException("渠道C支付失败", e);
            }
            return "渠道C支付成功，订单号：" + orderId;
        }, executor);
    }
    public static void main(String[] args) {
        String orderId = "ORDER_12345";
        long start = System.currentTimeMillis();
        // 1. 并行调用三个支付渠道
        CompletableFuture<String> payA = payByChannelA(orderId);
        CompletableFuture<String> payB = payByChannelB(orderId);
        CompletableFuture<String> payC = payByChannelC(orderId);
        // 2. 任意一个渠道完成即返回
        CompletableFuture<Object> anyFuture = CompletableFuture.anyOf(payA, payB, payC);
        // 3. 获取最快的支付结果
        String result = (String) anyFuture.join();
        System.out.println("最终支付结果：" + result);
        System.out.println("总耗时：" + (System.currentTimeMillis() – start) + "ms");
        executor.shutdown();
    }
}

运行结果：

渠道A处理支付：pool-1-thread-1
渠道B处理支付：pool-1-thread-2
渠道C处理支付：pool-1-thread-3
最终支付结果：渠道B支付成功，订单号：ORDER_12345
总耗时：520ms

核心结论：

四、任务依赖组合 API

除了串行和并行，还有一类常见场景：两个任务都完成后，执行一个依赖于这两个任务结果的新任务。比如 “查询用户信息” 和 “查询用户订单” 都完成后，聚合为 “用户完整数据”。CompletableFuture 提供了 thenCombine() 、 thenAcceptBoth() 等 API 来实现这种依赖组合。

1、thenCombine ()：组合两个任务结果并返回新结果

thenCombine() 方法接收两个参数：一个 CompletableFuture 和一个 BiFunction 函数式接口。它的核心逻辑是：当前任务和传入的任务都完成后，将两个任务的结果传入 BiFunction，执行并返回新的结果。

核心方法定义：

public <U,V> CompletableFuture<V> thenCombine(CompletableFuture<? extends U> other, BiFunction<? super T,? super U,? extends V> fn)
public <U,V> CompletableFuture<V> thenCombineAsync(CompletableFuture<? extends U> other, BiFunction<? super T,? super U,? extends V> fn)
public <U,V> CompletableFuture<V> thenCombineAsync(CompletableFuture<? extends U> other, BiFunction<? super T,? super U,? extends V> fn, Executor executor)

实战示例：查询用户信息 + 查询订单 → 聚合完整数据

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class ThenCombineDemo {
    private static final ExecutorService executor = Executors.newFixedThreadPool(2);
    static class User {
        private String userId;
        private String userName;
        public User(String userId, String userName) {
            this.userId = userId;
            this.userName = userName;
        }
        public String getUserId() {
            return userId;
        }
        public String getUserName() {
            return userName;
        }
    }
    static class Order {
        private String orderId;
        private String userId;
        public Order(String orderId, String userId) {
            this.orderId = orderId;
            this.userId = userId;
        }
        public String getOrderId() {
            return orderId;
        }
    }
    // 模拟查询用户信息
    private static CompletableFuture<User> queryUser(String userId) {
        return CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(600);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return new User(userId, "张三");
        }, executor);
    }
    // 模拟查询用户订单
    private static CompletableFuture<Order> queryOrder(String userId) {
        return CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(800);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return new Order("ORDER_123", userId);
        }, executor);
    }
    public static void main(String[] args) {
        String userId = "1001";
        // 任务1：查询用户信息
        CompletableFuture<User> userFuture = queryUser(userId);
        // 任务2：查询用户订单
        CompletableFuture<Order> orderFuture = queryOrder(userId);
        // 任务3：组合两个任务的结果，生成完整数据
        CompletableFuture<String> combinedFuture = userFuture.thenCombineAsync(orderFuture, (user, order) -> {
            return "用户完整数据：\\n" +
                    "用户ID：" + user.getUserId() + "\\n" +
                    "用户名：" + user.getUserName() + "\\n" +
                    "关联订单：" + order.getOrderId();
        }, executor);
        // 获取结果
        System.out.println(combinedFuture.join());
        executor.shutdown();
    }
}

运行结果：

用户完整数据：
用户ID：1001
用户名：张三
关联订单：ORDER_123

2、thenAcceptBoth ()：组合两个任务结果并消费

thenAcceptBoth() 与 thenCombine() 类似，但它无返回值，适用于 “组合两个任务结果并消费” 的场景（如打印、存储）。

核心方法定义：

public <U> CompletableFuture<Void> thenAcceptBoth(CompletableFuture<? extends U> other, BiConsumer<? super T,? super U> action)
public <U> CompletableFuture<Void> thenAcceptBothAsync(CompletableFuture<? extends U> other, BiConsumer<? super T,? super U> action)
public <U> CompletableFuture<Void> thenAcceptBothAsync(CompletableFuture<? extends U> other, BiConsumer<? super T,? super U> action, Executor executor)

实战示例：组合用户信息和订单并打印

// 基于上述示例的userFuture和orderFuture
userFuture.thenAcceptBoth(orderFuture, (user, order) -> {
    System.out.println("组合结果：用户" + user.getUserName() + "的订单是" + order.getOrderId());
}).join();

运行结果：

组合结果：用户张三的订单是ORDER_123

3、runAfterBoth ()：两个任务完成后执行无参任务

runAfterBoth() 是最简化的依赖组合 API，它不关心两个任务的结果，仅在两个任务都完成后执行一个无参数、无返回值的任务。

核心方法定义：

public CompletableFuture<Void> runAfterBoth(CompletableFuture<?> other, Runnable action)
public CompletableFuture<Void> runAfterBothAsync(CompletableFuture<?> other, Runnable action)
public CompletableFuture<Void> runAfterBothAsync(CompletableFuture<?> other, Runnable action, Executor executor)

实战示例：用户信息和订单查询完成后，记录操作日志

// 基于上述示例的userFuture和orderFuture
userFuture.runAfterBoth(orderFuture, () -> {
    System.out.println("用户信息和订单查询都已完成，记录日志");
}).join();

运行结果：

用户信息和订单查询都已完成，记录日志

五、关键注意点与避坑指南

1.Async 后缀的线程池选择

• 带 Async 的方法，若传入自定义线程池，则使用自定义线程池；否则使用默认的 ForkJoinPool.commonPool() 。

• 建议 始终使用自定义线程池 ，避免默认线程池的资源竞争问题（后续线程池篇会详细讲解）。

2.链式调用的结果传递机制

• 串行调用中，后续任务的参数是前序任务的返回值，形成一条 “结果传递链”。

• 若链式调用中某一个任务抛出异常，后续任务会直接终止。

3.避免任务依赖循环

• 不要在任务 A 的回调中依赖任务 B 的结果，同时在任务 B 的回调中依赖任务 A 的结果，这会导致 死锁 。

4.allOf () 与 anyOf () 的结果获取

• allOf()需手动调用每个子任务的 join() 或 get() 获取结果，注意捕获单个任务的异常。

• anyOf()返回的是 Object 类型，需根据实际场景进行类型转换，避免类型转换异常。

5.thenCompose () 的使用边界

• thenCompose()仅适用于 返回值为 CompletableFuture 的异步转换场景，若为同步转换，优先使用 thenApply() 。

• 避免滥用 thenCompose() ：如果前序任务和后续任务无依赖关系，应使用并行组合 API（如 allOf() ），而非串行的 thenCompose() 。

六、总结

本篇我们深入讲解了 CompletableFuture 的三大核心组合能力：

• 串行组合：通过 thenRun()、thenAccept()、thenApply() 实现任务的链式执行，重点区分 Async 后缀的线程差异；通过 thenCompose() 解决嵌套 CompletableFuture 的扁平化问题。

• 并行组合：通过 allOf() 等待所有任务完成，anyOf() 取最快完成的任务结果，满足多任务并行聚合的需求。

• 依赖组合：通过 thenCombine()、thenAcceptBoth() 实现两个任务的结果依赖组合。

掌握这些 API 后，你已经能够应对大部分异步编程场景。但在实际开发中， 异常处理是绕不开的话题 —— 如果链式调用中某个任务失败怎么办？ allOf() 中某个子任务异常如何降级？这些问题，我们将在下一篇：异常处理篇中详细解答。