thenCompose + 自定义 Executor + 超时

1. thenCompose vs. thenApply：区别与使用时机

在 Java 的异步编程（通过 CompletableFuture）中，经常需要执行一系列步骤。为此有两个相似的方法：thenApply 和 thenCompose。但它们的工作方式不同！

thenApply

当下一步只是对结果做一次简单转换、且不启动新的异步操作时，使用 thenApply。它接收上一步的结果，对其进行处理并返回一个新值（不是 CompletableFuture）。

如果你熟悉 Stream API，thenApply 的行为大致类似 map：取到结果、应用函数并返回转换后的值。

示例：

CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> "42");
CompletableFuture<Integer> lengthFuture = cf.thenApply(s -> s.length());
// lengthFuture 包含 2（字符串 "42" 的长度）

简单说，thenApply 就是在表达：“当结果准备好时，对它做这件事”。

thenCompose

• 用于下一步是另一个异步操作（返回 CompletableFuture）的场景。
• 可以“展开”嵌套的 CompletableFuture（类似 flatMap）。
• 如果对异步函数使用 thenApply，会得到 CompletableFuture<CompletableFuture<T>> —— 不方便！

示例：

CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> "user42");

// 假设我们需要根据用户名异步获取该用户的订单
CompletableFuture<List<Order>> ordersFuture = cf.thenCompose(username -> fetchOrdersAsync(username));
// fetchOrdersAsync 返回 CompletableFuture<List<Order>>

可视化：

• thenApply：CF<String> → thenApply(s -> s.length()) → CF<Integer>
• thenCompose：CF<User> → thenCompose(u -> fetchOrdersAsync(u.id)) → CF<List<Order>>

什么时候用哪个？

• 函数返回普通值 —— 使用 thenApply。
• 函数返回 CompletableFuture —— 使用 thenCompose。

错误示例：

cf.thenApply(username -> fetchOrdersAsync(username)); // 得到 CF<CF<List<Order>>>
cf.thenCompose(username -> fetchOrdersAsync(username)); // 得到 CF<List<Order>>

2. 线程池（Executor）管理：为何以及如何使用自定义 Executor

默认：ForkJoinPool.commonPool()

当你在没有指定 Executor 的情况下调用 CompletableFuture.supplyAsync(...) 或 thenApplyAsync(...) 时，Java 会使用公共线程池 —— ForkJoinPool.commonPool()。这通常很方便，但并非总是合适：

• 如果存在大量耗时或阻塞操作（网络请求、文件操作），公共池可能会“塞满”，所有任务都会等待。
• 有时需要隔离不同优先级的任务，或限制同时工作的线程数量。

何时需要自定义 Executor？

• 耗时、阻塞型操作（例如数据库、HTTP 请求、文件读取）。
• 任务隔离：避免用户任务影响系统任务。
• 资源限制：例如不同时启动超过 10 个下载。

如何创建自定义 Executor

通常使用 ThreadPoolExecutor 或 Executors 提供的工厂方法：

ExecutorService myExecutor = Executors.newFixedThreadPool(10);

如何在 CompletableFuture 中使用自定义 Executor

• 在 supplyAsync、runAsync、thenApplyAsync、thenComposeAsync 等方法中，可以传入第二个参数 —— 你的 Executor。

示例：

CompletableFuture<String> cf = CompletableFuture.supplyAsync(
    () -> loadDataFromNetwork(), myExecutor
);

cf.thenApplyAsync(data -> processData(data), myExecutor)
  .thenAcceptAsync(result -> System.out.println(result), myExecutor);

重要：如果不指定 Executor，将使用 ForkJoinPool.commonPool()。

何时默认 Executor 就足够？

• 用于短小的、CPU 计算型任务（简单计算）。
• 当不关心任务在哪个线程执行时。

3. 超时处理：orTimeout 与 completeOnTimeout

异步操作可能卡住或执行过久（例如服务器无响应）。为了避免“永远等待”，CompletableFuture 提供了处理超时的方法。

orTimeout

• 如果操作未在指定时间内完成，则以 TimeoutException 使 CompletableFuture 失败结束。
• 不会取消实际正在执行的任务，但下游链路会收到错误。

语法：

cf.orTimeout(3, TimeUnit.SECONDS)
  .exceptionally(ex -> {
      System.out.println("超时：" + ex);
      return null;
  });

示例：

CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> {
    Thread.sleep(5000); // 模拟耗时操作
    return "OK";
});

cf.orTimeout(2, TimeUnit.SECONDS)
  .exceptionally(ex -> {
      System.out.println("错误：" + ex);
      return "TIMEOUT";
  });

结果：

A TimeoutException will be thrown after 2 seconds and handled by exceptionally.

completeOnTimeout

• 如果操作未在超时时间内完成，则以指定的值完成 CompletableFuture。
• 不会抛出异常，而是返回一个“备用”值。

语法：

cf.completeOnTimeout("DEFAULT", 2, TimeUnit.SECONDS);

示例：

CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> {
    Thread.sleep(5000);
    return "OK";
});

cf.completeOnTimeout("TIMEOUT", 2, TimeUnit.SECONDS)
  .thenAccept(System.out::println); // 2 秒后会输出 "TIMEOUT"

orTimeout 与 completeOnTimeout 对比

orTimeout

completeOnTimeout

4. 实践：thenCompose、自定义 Executor 与超时示例

目标：

• 根据 id 异步获取用户（带延迟）。
• 然后异步获取该用户的订单列表（也带延迟）。
• 使用自定义的 Executor。
• 为获取订单添加超时。

import java.util.concurrent.*;
import java.util.*;

public class AsyncDemo {
    static ExecutorService ioExecutor = Executors.newFixedThreadPool(4);

    // 模拟异步获取用户
    static CompletableFuture<String> fetchUserAsync(int userId) {
        return CompletableFuture.supplyAsync(() -> {
            sleep(1000);
            return "user" + userId;
        }, ioExecutor);
    }

    // 模拟异步获取用户订单
    static CompletableFuture<List<String>> fetchOrdersAsync(String username) {
        return CompletableFuture.supplyAsync(() -> {
            sleep(3000); // 耗时操作！
            return List.of("order1", "order2");
        }, ioExecutor);
    }

    static void sleep(long ms) {
        try { Thread.sleep(ms); } catch (InterruptedException ignored) {}
    }

    public static void main(String[] args) {
        fetchUserAsync(42)
            .thenCompose(username ->
                fetchOrdersAsync(username)
                    .orTimeout(2, TimeUnit.SECONDS) // 获取订单的超时
                    .exceptionally(ex -> {
                        System.out.println("无法获取订单：" + ex);
                        return List.of();
                    })
            )
            .thenAccept(orders -> System.out.println("订单: " + orders))
            .join(); // 等待整条链路结束

        ioExecutor.shutdown();
    }
}

过程说明：

• 获取用户（1 秒）。
• 获取订单（3 秒，但超时为 2 秒）。
• 若未及时完成 —— 捕获 TimeoutException，返回空列表。
• 全部运行在自定义的 Executor 中。

结果：

无法获取订单：java.util.concurrent.TimeoutException
订单: []

如果将 fetchOrdersAsync 中的延迟减小到 1_000 毫秒 —— 就会看到真实的订单。

5. 常见错误与注意事项

错误 1：使用 thenApply 而不是 thenCompose 来处理异步操作。
如果函数返回 CompletableFuture，却使用了 thenApply，会得到嵌套类型 CompletableFuture<CompletableFuture<T>>。这会让链路变复杂并产生多余的包装。解决方案：使用 thenCompose 将结果“扁平化”为 CompletableFuture<T>。

错误 2：在没有自定义 Executor 的情况下运行耗时或 IO 任务。
默认情况下任务运行在 ForkJoinPool.commonPool()。如果它被压满，延迟会增长，应用中的其他任务也可能变慢。解决方案：创建自己的 ExecutorService，并将其传入 supplyAsync/thenApplyAsync。

错误 3：误以为 orTimeout 会取消任务的执行。
orTimeout 只是让 CompletableFuture 因超时而失败，但任务本身仍在后台运行。解决方案：若需要停止执行，使用 cancel(true) 或自定义的中断机制。

错误 4：误解超时的作用范围。
orTimeout 和 completeOnTimeout 只作用于链路中的某一个具体步骤，而非整个链路。解决方案：如果需要对整条链路设置统一超时，将其包装到一个单独的 CompletableFuture 上并对其应用超时。

错误 5：未关闭 ExecutorService。
如果任务结束后没有调用 shutdown()/shutdownNow()，线程会继续存活，程序可能“挂住”。解决方案：始终在 finally 中关闭 ExecutorService，或在 Java 21+ 使用 try-with-resources。

1

调查/小测验

异步编程第 55 级，课程 4

不可用

异步编程

异步编程

打开