Java 高并发实战：线程池调优 任务超时控制 CompletableFuture 编排

Java 高并发实战：线程池调优 × 任务超时控制 × CompletableFuture 编排

在现代高并发、高性能的 Java 应用中，有效地管理异步任务和执行流程是至关重要的。Java 并发包 (java.util.concurrent) 提供了强大的工具集，其中 线程池 (ThreadPoolExecutor) 和 CompletableFuture 是两大核心利器。本文将深入探讨如何结合使用它们来实现任务的异步执行、超时控制以及复杂流程编排，并给出实战中的最佳实践。

第一部分：线程池 (ThreadPoolExecutor) 基础

线程池是一种池化技术，用于管理线程的生命周期，避免频繁创建和销毁线程带来的性能开销。

1.1 创建线程池

推荐使用 ThreadPoolExecutor 构造函数来创建，能够提供更细粒度的控制。

import java.util.concurrent.*;
 
ThreadFactory namedThreadFactory = new ThreadFactoryBuilder()
        .setNameFormat("custom-pool-%d") // 自定义线程名称
        .setDaemon(true)
        .build();
 
ThreadPoolExecutor customThreadPool = new ThreadPoolExecutor(
    4, // 核心线程数
    10, // 最大线程数
    60L, TimeUnit.SECONDS, // 空闲线程存活时间
    new LinkedBlockingQueue<>(100), // 工作队列
    namedThreadFactory, // 自定义线程工厂
    new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略
);

关键参数解析：

1.2 提交任务

// Runnable，无返回值
customThreadPool.execute(() -> {
    System.out.println("异步任务执行中...");
});
 
// Callable，有返回值
Future<String> future = customThreadPool.submit(() -> {
    Thread.sleep(1000);
    return "任务结果";
});

第二部分：异步任务超时控制

在实际应用中，我们经常需要对一个可能长时间运行的任务设置超时，防止其阻塞主流程或耗尽资源。

2.1 Future.get() 超时控制

ExecutorService executor = Executors.newFixedThreadPool(2);
 
Future<String> future = executor.submit(() -> {
    Thread.sleep(4000); // 耗时任务
    return "Success";
});
 
try {
    String result = future.get(2, TimeUnit.SECONDS); // 2 秒超时
    System.out.println("结果: " + result);
} catch (TimeoutException e) {
    System.err.println("任务超时，取消执行！");
    future.cancel(true); // 中断任务
}

2.2 批量任务超时控制 (invokeAll)

List<Callable<String>> tasks = Arrays.asList(
    () -> { Thread.sleep(3000); return "Task1"; },
    () -> { Thread.sleep(1000); return "Task2"; }
);
 
ExecutorService executor = Executors.newFixedThreadPool(2);
List<Future<String>> futures = executor.invokeAll(tasks, 3, TimeUnit.SECONDS);

for (Future<String> f : futures) {
    if (f.isDone()) {
        System.out.println("结果: " + f.get());
    } else {
        System.out.println("任务超时未完成");
    }
}

第三部分：CompletableFuture 任务编排

CompletableFuture 提供了丰富的 API，用于组合、编排多个异步任务。

3.1 创建 CompletableFuture

CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> "来自 commonPool");
 
CompletableFuture<String> cf2 = CompletableFuture.supplyAsync(() -> "来自自定义线程池",
        customThreadPool);

3.2 链式调用与结果转换

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello")
    .thenApply(s -> s + " World")
    .thenApply(String::toUpperCase);
 
System.out.println(future.join()); // HELLO WORLD

3.3 任务组合

// thenCompose：依赖关系
CompletableFuture<String> getUser = CompletableFuture.supplyAsync(() -> "用户ID:123");
CompletableFuture<String> getOrder = getUser.thenCompose(userId ->
    CompletableFuture.supplyAsync(() -> userId + " 的订单")
);
 
// thenCombine：合并结果
CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> "Hello");
CompletableFuture<String> f2 = CompletableFuture.supplyAsync(() -> "World");
CompletableFuture<String> combined = f1.thenCombine(f2, (s1, s2) -> s1 + " " + s2);

3.4 并行执行 (allOf, anyOf)

CompletableFuture<String> t1 = CompletableFuture.supplyAsync(() -> "结果1");
CompletableFuture<String> t2 = CompletableFuture.supplyAsync(() -> "结果2");
CompletableFuture<String> t3 = CompletableFuture.supplyAsync(() -> "结果3");
 
// allOf
CompletableFuture<Void> all = CompletableFuture.allOf(t1, t2, t3);
all.thenRun(() -> System.out.println(Arrays.asList(t1.join(), t2.join(), t3.join())));
 
// anyOf
CompletableFuture.anyOf(t1, t2, t3)
    .thenAccept(result -> System.out.println("最先完成: " + result));

3.5 异常处理

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    if (Math.random() > 0.5) throw new RuntimeException("异常！");
    return "Success";
});
 
future.exceptionally(ex -> {
    System.err.println("异常: " + ex.getMessage());
    return "默认值";
}).thenAccept(System.out::println);

3.6 超时控制

CompletableFuture<String> slowTask = CompletableFuture.supplyAsync(() -> {
    try { Thread.sleep(5000); } catch (InterruptedException ignored) {}
    return "最终结果";
});
 
// 超时抛异常
slowTask.orTimeout(2, TimeUnit.SECONDS);
 
// 超时返回默认值
slowTask.completeOnTimeout("默认值", 2, TimeUnit.SECONDS);

ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
CompletableFuture<String> promise = new CompletableFuture<>();
 
customThreadPool.submit(() -> {
    try {
        promise.complete(doWork()); // 正常完成
    } catch (Exception e) {
        promise.completeExceptionally(e);
    }
});
 
scheduler.schedule(() -> {
    if (!promise.isDone()) {
        promise.completeExceptionally(new TimeoutException("超时！"));
    }
}, 2, TimeUnit.SECONDS);

第四部分：实战技巧与最佳实践

System.out.printf("PoolSize=%d, Active=%d, Queue=%d%n",
    customThreadPool.getPoolSize(),
    customThreadPool.getActiveCount(),
    customThreadPool.getQueue().size()
);

总结

通过熟练结合 线程池、Future 超时控制 和 CompletableFuture 编排能力，我们可以构建出 高效、健壮、可维护 的异步应用程序。在实际项目中，结合自定义线程池命名、超时保护、非阻塞编排与线程池监控，可以显著提升系统的稳定性和可观测性。