14.3 线程池与任务结果

学完线程创建与同步、通信和死锁后，再用约 90 分钟完成这篇 Java 17 实验。你会配置有界线程池，观察任务失败与拒绝，并按阶段关闭执行器。

工位与等候区都有限

线程池解决三件事：复用工作线程、限制并发执行量、管理任务积压。只设置线程数而留下无界队列，仍然没有完成容量设计。

1. 从线程转向执行器

java

ExecutorService pool = Executors.newFixedThreadPool(4);
try {
    Future<String> future = pool.submit(() -> "catalog");
    String result = future.get();
} finally {
    pool.shutdown();
}

Executor 只定义 execute(Runnable)，把提交与执行分开；
ExecutorService 增加 submit、批量任务、关闭和终止等待；
ThreadPoolExecutor 暴露工作线程、队列、工厂和拒绝策略等容量参数；
ScheduledExecutorService 处理延迟与周期任务。

业务对象通常依赖 Executor 或 ExecutorService 接口；应用装配层负责创建和关闭具体线程池。

2. `ThreadPoolExecutor` 怎样接收任务

java

ThreadPoolExecutor pool = new ThreadPoolExecutor(
        2,                               // corePoolSize
        4,                               // maximumPoolSize
        30, TimeUnit.SECONDS,            // 非核心空闲线程存活时间
        new ArrayBlockingQueue<>(100),   // 有界等候区
        threadFactory,
        new ThreadPoolExecutor.AbortPolicy());

执行器仍在运行、线程工厂能创建线程时，提交过程大致如下；并发关闭会让已经尝试入队的任务也被拒绝：

当前工作线程少于 corePoolSize，创建核心线程；
否则尝试把任务放进工作队列；
队列满且线程数少于 maximumPoolSize，再创建线程；
队列满、线程也到上限，或执行器已经关闭，交给拒绝策略。

一个常见误解是“只要设置了 maximumPoolSize，池就会扩容”。若队列无界，任务通常一直入队，线程数不会因队列满而超过核心数。

3. 线程工厂让工作线程可识别

java

AtomicInteger number = new AtomicInteger();
ThreadFactory factory = task -> {
    Thread thread = new Thread(task,
            "library-worker-" + number.incrementAndGet());
    thread.setDaemon(false);
    thread.setUncaughtExceptionHandler((t, error) ->
            System.err.println(t.getName() + " 未捕获异常：" + error));
    return thread;
};

线程名会出现在日志、监控和线程转储中。普通业务池通常使用非守护线程，并由拥有者显式关闭；不要指望 JVM 退出时替守护线程完成数据刷新。

线程工厂负责线程创建；请求级上下文应在每次任务执行前安装，并在 finally 中恢复或清理。只在工厂设置一次安全上下文、日志上下文或 ThreadLocal，无法适应工作线程处理不同请求。

4. 队列与拒绝策略表达过载政策

策略	行为	风险与适用边界
`AbortPolicy`	抛 `RejectedExecutionException`	默认；调用者必须处理或向上报告
`CallerRunsPolicy`	提交线程同步执行任务	可减慢提交端，但会抬高调用延迟；关闭后不会执行任务
`DiscardPolicy`	静默丢弃新任务	只有业务明确允许丢弃且已有指标时才考虑
`DiscardOldestPolicy`	丢掉队首，再尝试提交	破坏顺序；优先级队列中“最老”语义还可能更危险

拒绝不是线程池故障，而是容量边界开始发挥作用。系统要明确选择：

给上游返回“繁忙，请重试”；
让提交者限速或阻塞；
降级到更便宜的工作；
将任务写入具备持久化和容量治理的外部队列；
对确实可丢弃的数据记录丢弃计数。

CallerRunsPolicy 还意味着 maximumPoolSize 只限制池内工作线程，提交者可能额外并发执行任务。丢弃策略和关闭后的 CallerRunsPolicy 不会自动取消被丢弃的 FutureTask；submit() 可能返回永远不完成的 Future，调用方必须有明确的取消或拒绝处理。

不要用无限等待把过载藏起来。队列深度、活跃线程、拒绝数、任务等待时间和执行时间都应进入监控。

5. 快捷工厂不是一律不能用

工厂	实现特征	使用条件
`newFixedThreadPool(n)`	固定线程数，无界 `LinkedBlockingQueue`	提交量在外部已严格有界
`newSingleThreadExecutor()`	单工作线程，无界队列	需要顺序，且积压有外部上限
`newCachedThreadPool()`	`SynchronousQueue`，需要时创建线程	短任务，并发量已受外部约束
`newScheduledThreadPool(n)`	延迟工作队列	定时任务有失败、重叠和关闭策略

“生产环境禁止 Executors”过于绝对。真正的问题是默认容量是否符合系统约束。若不能证明积压有界，就显式使用 ThreadPoolExecutor 和有界队列。

6. 线程数从测量开始

CPU 密集型任务可以从接近 Runtime.getRuntime().availableProcessors() 的线程数开始；大量等待 I/O 的任务可能允许更多线程。两者都只是压测起点。

决定线程数时还要看：

单任务 CPU 时间与等待时间；
目标吞吐和尾延迟；
每线程栈与任务对象的内存；
数据库连接池、远端限流等下游容量；
容器 CPU 配额和 JVM 实际可见核心；
任务是否会持锁、嵌套提交或阻塞等待同一线程池。

“CPU 核数 + 1”或“核数 × 2”不能作为通用生产结论。先给资源设上限，再用代表性流量测量。

7. `Runnable`、`Callable` 与 `Future`

Runnable.run() 不返回结果，也不声明受检异常；Callable<V>.call() 返回 V，并允许抛出受检异常。以下片段假设已有 Book 和 catalog，所在方法继续传播 InterruptedException；实际业务还需要把超时和失败转成明确结果。

java

Future<Book> future = pool.submit(() -> catalog.load("BK-14"));

try {
    Book book = future.get(500, TimeUnit.MILLISECONDS);
} catch (TimeoutException e) {
    future.cancel(true);
} catch (ExecutionException e) {
    Throwable taskFailure = e.getCause();
}

execute(Runnable) 没有结果句柄；未捕获异常走工作线程的未捕获异常路径，工作线程可能被池替换；
submit(...) 返回 Future，任务抛出的异常由 get() 以 ExecutionException 报告；
成功的 get() 建立任务完成到调用者的可见性关系；
get(timeout, unit) 超时只停止等待，不自动停止任务；
cancel(true) 请求中断正在运行的任务，任务仍要传播或检查中断；
cancel(false) 不请求中断已经运行的任务；取消成功后，后续 get() 抛 CancellationException，即使底层代码还没退出。

注意，CallerRunsPolicy 可能让 submit() 自身执行任务并阻塞，后面的 get(timeout) 并不限制这段时间。

如果提交后丢掉 Future，submit() 的失败很容易变成“静默失败”。无返回值任务也应有统一的失败记录方式。

实验：有界池、任务结果、异常与关闭

java

import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class ExecutorDemo {
    public static void main(String[] args) throws Exception {
        AtomicInteger number = new AtomicInteger();
        ThreadFactory factory = task -> {
            Thread thread = new Thread(task, "library-worker-" + number.incrementAndGet());
            thread.setDaemon(false);
            return thread;
        };
        ThreadPoolExecutor pool = new ThreadPoolExecutor(
                2,
                2,
                0L,
                TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(4),
                factory,
                new ThreadPoolExecutor.CallerRunsPolicy());

        boolean terminated;
        try {
            List<Future<String>> futures = List.of(
                    pool.submit(() -> "catalog"),
                    pool.submit(() -> "receipt"),
                    pool.submit(() -> "audit"));
            List<String> results = futures.stream().map(ExecutorDemo::get).toList();
            System.out.println("任务结果：" + results);

            Future<String> failed = pool.submit(() -> {
                throw new IllegalStateException("库存快照损坏");
            });
            try {
                failed.get();
                throw new AssertionError("失败任务不应返回结果");
            } catch (ExecutionException e) {
                System.out.println("任务异常：" + e.getCause().getClass().getSimpleName());
            }
        } finally {
            pool.shutdown();
            terminated = awaitTermination(pool);
        }

        System.out.println("已终止：" + terminated);
        if (!terminated) {
            throw new AssertionError("线程池没有终止");
        }
        System.out.println("线程池检查通过");
    }

    private static String get(Future<String> future) {
        try {
            return future.get();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException("等待任务时被中断", e);
        } catch (ExecutionException e) {
            throw new IllegalStateException("任务失败", e.getCause());
        }
    }

    private static boolean awaitTermination(ThreadPoolExecutor pool) {
        try {
            if (pool.awaitTermination(2, TimeUnit.SECONDS)) {
                return true;
            }
            pool.shutdownNow();
            return pool.awaitTermination(2, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            pool.shutdownNow();
            Thread.currentThread().interrupt();
            return false;
        }
    }
}

运行：

bash

chapter14_executor_root=$(mktemp -d)
mkdir -p "$chapter14_executor_root/out"
# 将上面的代码保存为 "$chapter14_executor_root/ExecutorDemo.java"
javac --release 17 -Xlint:all -Werror \
  -d "$chapter14_executor_root/out" \
  "$chapter14_executor_root/ExecutorDemo.java"
java -ea -cp "$chapter14_executor_root/out" ExecutorDemo
rm -rf -- "$chapter14_executor_root"

输出：

text

任务结果：[catalog, receipt, audit]
任务异常：IllegalStateException
已终止：true
线程池检查通过

三个任务可能按任意顺序运行，程序却按 Future 列表顺序取值，所以结果列表稳定。失败任务必须调用 get() 才观察到包装后的原因。

队列容量是 4，上面的实验不会触发拒绝。把下面的独立实验保存为 SaturationDemo.java，使用相同的 Java 17 编译参数运行。两个工作线程被闩锁占住，一个任务排队，因此第四个任务确定由提交者执行。最后也验证关闭后提交会留下待取消的 Future：

java

import java.util.concurrent.*;

public class SaturationDemo {
    public static void main(String[] args) throws Exception {
        ThreadPoolExecutor pool = new ThreadPoolExecutor(2, 2, 0, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(1), new ThreadPoolExecutor.CallerRunsPolicy());
        CountDownLatch ready = new CountDownLatch(2);
        CountDownLatch release = new CountDownLatch(1);
        Runnable held = () -> {
            ready.countDown();
            try {
                release.await();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        };
        try {
            pool.execute(held);
            pool.execute(held);
            ready.await();
            pool.execute(() -> {}); // One queued task; both workers remain held.
            Thread caller = Thread.currentThread();
            Future<Thread> fourth = pool.submit(Thread::currentThread);
            if (fourth.get() != caller) throw new AssertionError("Expected caller execution");
            System.out.println("Caller executed fourth task: true");
        } finally {
            release.countDown();
            pool.shutdown();
            if (!pool.awaitTermination(2, TimeUnit.SECONDS)) {
                pool.shutdownNow();
                throw new AssertionError("Pool did not terminate");
            }
        }
        Future<?> discarded = pool.submit(() -> {});
        if (discarded.isDone()) throw new AssertionError("Unexpected completed future");
        discarded.cancel(false); // CallerRunsPolicy discarded this task after shutdown.
        System.out.println("Discarded future explicitly cancelled: " + discarded.isCancelled());
    }
}

text

Caller executed fourth task: true
Discarded future explicitly cancelled: true

8. 分阶段关闭

shutdown() 不再接收新任务，但会处理已经提交的任务。shutdownNow() 尝试中断正在运行的任务，并返回尚未开始的队列任务；它不是“确保所有线程立即消失”。

推荐流程：

调用 shutdown()；
在业务允许的期限内 awaitTermination()；
超时则调用 shutdownNow()；
再等待一个短期限并记录未终止；
当前线程被中断时，调用 shutdownNow()，恢复中断状态并尽快返回。

shutdownNow() 返回的排队任务不会因此自动进入 Future 的取消状态；持有这些 Future 的组件还应明确取消它们或通知等待者。Java 17 的 ExecutorService 还不是 AutoCloseable，本篇用 finally 管理关闭。

线程池应由创建它的组件关闭。不要让每个借阅服务方法各自关闭共享池，也不要在每个请求中创建一个新池。

9. 周期任务与 Fork/Join

调度任务

java

ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
scheduler.scheduleAtFixedRate(task, 0, 1, TimeUnit.MINUTES);

scheduleAtFixedRate 以计划开始时间为基准；任务太慢时不会让同一周期任务并行重入，但可能连续追赶；
scheduleWithFixedDelay 在上次完成后再等待固定间隔；
周期任务抛出未捕获异常后，后续执行会被抑制；要明确记录和恢复政策；
仍然需要关闭调度器。

Fork/Join

ForkJoinPool 使用工作窃取，适合能递归拆成较小子任务的 CPU 计算。它不是所有任务的“更快线程池”。

常见 OpenJDK 实现会让并行流使用 common pool，但 Stream API 不承诺固定的执行器。共享池中的长阻塞任务可能影响其他并行流和 CompletableFuture 默认异步任务。需要隔离或容量上限时，显式向执行器提交任务；不要假定并行流天然属于某个自建池。

10. 运维检查表

维度	应回答的问题
所有权	谁创建、谁关闭这个池？
容量	最大工作线程和最大排队任务是多少？
过载	队列满时拒绝、降级、阻塞还是持久化？
延迟	任务排队时间和执行时间分别是多少？
失败	`execute` 与 `submit` 的异常在哪里记录？
取消	任务是否响应中断，阻塞调用是否有超时？
下游	数据库连接、远程服务能承受多少并发？
上下文	`ThreadLocal`、日志和安全上下文如何清理？
关闭	优雅等待期限与强制取消策略是什么？

11. 常见故障

现象	可能原因
`maximumPoolSize` 从未生效	队列无界，从未进入扩容步骤
内存持续上涨	无界队列积压任务及其捕获对象
线程数迅速增加	cached pool 或自定义上限过大
`submit()` 任务失败但日志没有错误	丢弃了 `Future`，未调用 `get()`
主方法结束，进程仍在	非守护工作线程仍存活，池未关闭
`shutdownNow()` 后任务仍运行	任务吞中断或调用不可中断 API
`CallerRunsPolicy` 导致请求变慢	提交线程正在承担过载任务，这是其背压方式
池内任务互相等待后卡死	线程饥饿死锁；任务同步等待同一小池中的子任务

12. 练习

把实验队列容量改成 1，用闩锁稳定占住两个工作线程和一个队列位置，验证下一次提交执行 CallerRunsPolicy。断言执行线程名，而不是依赖时间。
给一个任务增加 100 毫秒超时。超时后调用 cancel(true)，让任务在捕获 InterruptedException 后恢复状态并退出。
分别用 execute() 和 submit() 运行一个抛异常的任务，记录异常到达的位置。
为图书馆写一页容量说明：峰值提交速率、单任务时间、队列上限、拒绝响应、下游连接上限和关闭期限。

关门以前，验证容量合同

把实验改成两个工作线程、一个队列位置，用闩锁稳定占满工位和队列，再提交第四个任务。若采用 CallerRunsPolicy，第四个任务应在提交线程执行；测试线程名，不要用休眠猜线程池已经饱和。

分别通过 execute 和 submit 运行失败任务：前者走工作线程的未捕获异常路径，后者把失败保存在 Future 中并由 get() 包装成 ExecutionException。提交后丢弃 Future，就等于放弃这条失败通道。

最后执行分阶段关闭：shutdown、限时等待、必要时 shutdownNow、再次等待。若当前线程被中断，先请求取消，再恢复中断状态。线程数、队列上限、拒绝政策、下游容量和关闭期限合在一起，才是一份完整的执行器合同。

查阅 Java 17 的正式契约

ThreadPoolExecutor：任务接收顺序、队列、线程工厂、拒绝和钩子。
ExecutorService：提交、关闭、终止与内存一致性效果。
Future：结果、异常、取消和 get 的可见性保证。
ScheduledThreadPoolExecutor：延迟与周期任务的执行规则。

继续学习：I/O 与文件

线程池把线程数、等待队列、拒绝行为与关闭期限放在同一处管理。下一章进入文件与目录；I/O 会阻塞，也最需要这里的容量、超时、中断和关闭原则。

继续学习 I/O 流与持久化边界

第1章逻辑

第2章集合论

第3章证明方法

第4章递归、归纳与递推

第5章关系与函数

第6章组合数学

第7章图论

第8章概率与统计

第9章数论

第10章线性代数

第11章微积分与优化

第12章信息论与数值计算

14.3 线程池与任务结果

工位与等候区都有限

1. 从线程转向执行器

2. `ThreadPoolExecutor` 怎样接收任务

3. 线程工厂让工作线程可识别

4. 队列与拒绝策略表达过载政策

5. 快捷工厂不是一律不能用

6. 线程数从测量开始

7. `Runnable`、`Callable` 与 `Future`

实验：有界池、任务结果、异常与关闭

8. 分阶段关闭

9. 周期任务与 Fork/Join

调度任务

Fork/Join

10. 运维检查表

11. 常见故障

12. 练习

关门以前，验证容量合同

查阅 Java 17 的正式契约

继续学习：I/O 与文件