《Java 网络编程基础》12.2 业务线程池与阻塞隔离

本节目标：把会阻塞的业务处理从事件循环挪到有界线程池，并设计一条可观测、可拒绝、能安全回到事件循环的路径。
适用版本：JDK 25（Temurin 25.0.5+8）；ThreadPoolExecutor、ArrayBlockingQueue、虚拟线程。

承接的问题

12.1 的结论很直接：事件循环线程上任何阻塞都会让所有连接和定时器一起等。可 linkd 的业务不可能都是纯内存计算。GET /books/{id} 要访问 libraryd，鉴权要查缓存或数据库，响应要序列化；这些动作即使只花 20 毫秒，乘上一千条并发连接，也会把循环压成一条长队。

6.2 的错误一已经点过方向：阻塞业务移到业务线程池。但「移过去」只是开始，后面有三条边界必须同时成立：

1. 业务线程不能碰 Connection、SelectionKey 或任何循环本地状态，写回必须 loop.execute(...)。
2. 事件循环不能等业务结果，Future.get()、CompletableFuture.join() 都会把阻塞带回循环。
3. 池子必须有界，队列满时要有明确的拒绝路径；否则只是把 OOM 从循环队列挪到了业务队列。

本机口径：Apple M1 Pro，10 核，32 GB，macOS 26.3 (25D125)；Temurin 25.0.5+8。下面的线程池与虚拟线程数字都来自这台机器，单次运行，未做统计。

线程池大小：先算等待比，再压测

线程数不是「CPU 核数乘二」这种固定口诀。Brian Goetz 的估算式是：

N_threads = N_cpu × U_cpu × (1 + W / C)

N_cpu 是可用核数，U_cpu 是目标 CPU 利用率，W 是任务等待外部资源的时间，C 是真正占用 CPU 的时间。linkd 的一个请求如果 CPU 工作 5ms、等待 libraryd 45ms，W/C = 9；10 核、目标 80% 利用率，算出来是 10 × 0.8 × 10 = 80 条线程。纯 CPU 任务 W/C = 0，只需要 8 条左右。

这个公式给的是量级，不是精确值。等待时间会随下游抖动，CPU 利用率也不是恒定值；最终要用压测的 p99 延迟和队列等待时间校正。另一个约束是 Little 定律：并发数 = 吞吐 × 延迟。目标 500 请求/秒、单请求 50ms，系统里同时在处理的请求大约是 25 个；池子小于这个数，请求就开始排队。

还要避开 ForkJoinPool.commonPool()。CompletableFuture.supplyAsync() 不传 executor 时就用它，并行度默认是「核数减一」，而且被所有未指定池的代码共享。只要有一个阻塞任务占住 commonPool，其他依赖它的任务就一起饿着；在 10 核机器上，这等于把业务并发压到 9 条线程。linkd 的异步编排一律显式传 business，不用默认池。

linkd 只设一个业务池，处理 HTTP 路由、鉴权、编码这些混合任务。纯 CPU 的离线任务不要塞进来：它们会占满线程，让等待型请求排在后面。需要时另开一个 N_cpu + 1 的 CPU 池，两个池子的队列和指标分开看。

ThreadPoolExecutor 用固定大小、prestartAllCoreThreads() 预热，避免第一批请求承担建线程的开销。不要用 Executors.newCachedThreadPool()：它的线程数没有上限，队列满就继续建线程，下游一慢，线程数会跟着请求数一起涨。

有界队列与拒绝策略

队列长度决定「能吸收多大的突发」。ArrayBlockingQueue 有界、数组实现、缓存友好；LinkedBlockingQueue 默认容量是 Integer.MAX_VALUE，Executors.newFixedThreadPool() 用的就是它，队列可以一直涨到 OOM。linkd 明确用 ArrayBlockingQueue，容量从「线程数的 2 到 4 倍」起步，再按 p99 排队时间调整。

队列满时，RejectedExecutionHandler 决定行为：

CallerRunsPolicy 是这里最危险的选项。它的语义是「池子忙，就让提交者自己执行」；如果提交者是事件循环线程，这个任务就会在循环上跑完，12.1 的阻塞问题原样回来，而且更隐蔽：压测低并发时永远触发不到。linkd 的自定义处理器只做两件事：rejected 计数加一，然后抛出 RejectedExecutionException；调用方捕获后，用 loop.execute 回一个 503。

拒绝不是失败，是背压信号。请求在入口被快速拒绝，比在无界队列里等 30 秒再超时更好：客户端能早点重试，服务端的内存和延迟都可控。14.1 会把「拒绝前先看读写水位」做成更细的背压，这里先把「有界 + 明确拒绝」立住。

虚拟线程对照：能替代什么

JDK 25 的虚拟线程已经正式可用（JEP 444 在 JDK 21 交付，后续版本持续修正）。Executors.newVirtualThreadPerTaskExecutor() 为每个任务创建一个虚拟线程，阻塞时由 JVM 把载体线程让出来，所以「一个请求一个线程」的阻塞式代码不再需要为每个请求保留一个平台线程。

它适合 linkd 的哪一部分？适合「业务代码是阻塞式、并发量高、每个任务大部分时间在等下游」的场景，比如代理 libraryd、查缓存、发通知。它不适合 CPU 密集任务：虚拟线程不会让 10 个核变成 20 个核，计算任务仍然受限于载体线程数。它也不自动提供有界并发：每个请求开一个虚拟线程，如果下游数据库只能承受 50 个连接，你需要额外的 Semaphore 限流，否则只是把压力更快地打到下游。

JDK 24 起，synchronized 块不再普遍把虚拟线程钉在载体线程上（JEP 491），但本地方法或外部函数调用仍可能造成钉住。不要把「虚拟线程」当成「所有阻塞都免费」；有 native 调用或旧库时，仍然要用 JFR 的虚拟线程事件确认。

同负载对照：200 个任务，每个睡眠 20ms，平台池 4 线程、队列 256，对比每任务一个虚拟线程。

export JAVA_HOME=~/.jdks/jdk-25.0.5+8/Contents/Home
javac --release 25 -d out $(find src -name '*.java')
java -cp out com.plume.linkd.core.BusinessPoolDemo

本机输出（单次运行，未做统计）：

platform submitted=200 completed=200 rejected=0 queue=0 active=0 threads=4 callerRejected=0
platform elapsed=1232 ms
virtual elapsed=30 ms
platform submitted=50 completed=6 rejected=44 queue=0 active=0 threads=2 callerRejected=44
burst elapsed=313 ms

200 个 20ms 任务在 4 条线程上是 50 轮，理论下限 1000ms，实测 1232ms，差额来自建线程、调度和 JVM 启动。虚拟线程把 200 个睡眠并发起来，30ms 就结束；这不是「虚拟线程更快」，而是它把「等待」从线程占用里拿掉了。第三组是 2 条线程、队列容量 4、50 个 100ms 任务：最多接收 2 + 4 = 6 个，其余 44 个立即被拒绝，耗时约 313ms 是这 6 个任务分三轮跑完的时间。

这些数字只说明机制，不是容量规划依据。虚拟线程在真实下游上的表现，取决于连接池、DNS、TLS 和下游限流；要用 15.2 的压测口径重新量。

在 linkd 里落地：BusinessExecutor

BusinessExecutor 封装 ThreadPoolExecutor，把「有界队列、命名线程、拒绝计数、指标快照、优雅关闭」放在一个类里。包名 com.plume.linkd.core。

package com.plume.linkd.core;

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

public final class BusinessExecutor implements AutoCloseable {
    private final ThreadPoolExecutor pool;
    private final LongAdder submitted = new LongAdder();
    private final LongAdder completed = new LongAdder();
    private final LongAdder rejected = new LongAdder();

    public BusinessExecutor(int threads, int queueCapacity, String namePrefix) {
        if (threads <= 0) throw new IllegalArgumentException("threads must be positive");
        if (queueCapacity <= 0) throw new IllegalArgumentException("queueCapacity must be positive");
        this.pool = new ThreadPoolExecutor(
                threads, threads, 0L, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(queueCapacity),
                new NamedThreadFactory(namePrefix),
                (task, executor) -> {
                    rejected.increment();
                    throw new RejectedExecutionException(
                            "business queue full, depth=" + executor.getQueue().size());
                });
        this.pool.prestartAllCoreThreads();
    }

    public void execute(Runnable task) {
        if (task == null) throw new NullPointerException("task");
        submitted.increment();
        pool.execute(() -> {
            try {
                task.run();
            } finally {
                completed.increment();
            }
        });
    }

    public Stats stats() {
        return new Stats(submitted.sum(), completed.sum(), rejected.sum(),
                pool.getQueue().size(), pool.getActiveCount(), pool.getPoolSize());
    }

    @Override
    public void close() {
        pool.shutdown();
        try {
            if (!pool.awaitTermination(5, TimeUnit.SECONDS)) pool.shutdownNow();
        } catch (InterruptedException e) {
            pool.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }

    public record Stats(long submitted, long completed, long rejected,
                        int queueDepth, int activeThreads, int poolThreads) {
        public String line() {
            return "submitted=" + submitted + " completed=" + completed
                    + " rejected=" + rejected + " queue=" + queueDepth
                    + " active=" + activeThreads + " threads=" + poolThreads;
        }
    }

    private static final class NamedThreadFactory implements ThreadFactory {
        private final String prefix;
        private final AtomicInteger sequence = new AtomicInteger();

        private NamedThreadFactory(String prefix) {
            this.prefix = prefix;
        }

        @Override
        public Thread newThread(Runnable task) {
            Thread thread = new Thread(task, prefix + sequence.getAndIncrement());
            thread.setDaemon(false);
            return thread;
        }
    }
}

几个设计点：

• 固定大小加预热。 corePoolSize == maximumPoolSize，线程数不会随负载膨胀；prestartAllCoreThreads() 把建线程成本放到启动阶段。
• 拒绝处理器在提交线程执行。 它只加计数并抛异常，不做日志刷盘、不做阻塞清理；否则拒绝路径本身会变成新的阻塞点。
• 计数用 LongAdder。 多个业务线程同时加，LongAdder 比 AtomicLong 少争用；stats() 是快照，读的时候不需要停世界。
• close() 先 shutdown() 再等 5 秒。 等在途任务结束；超时后 shutdownNow() 发中断。中断只对响应 InterruptedException 的代码有效，阻塞在 socket 读上的任务未必停得下来，所以关闭顺序要和 14.3 的连接排空配合。
• 线程命名带前缀。 biz-0、biz-1 在 jstack、JFR 和日志里能一眼区分事件循环线程与业务线程。

Stats 里的 queueDepth 和 activeThreads 是瞬时值，适合看趋势，不适合算准确吞吐；submitted - completed - rejected 才是当前在池内（排队 + 执行）的任务数。真正要报警的是拒绝率和排队时间：拒绝率持续大于零，说明容量或下游有问题；排队时间 p99 上涨，说明线程数不够或任务变慢。

排队时间可以在提交时打点、任务开始时结算。下面的片段给的是平均值，生产上应换成固定桶计数，否则长尾会被平均值吃掉：

private final LongAdder queueWaitNanos = new LongAdder();

public void execute(Runnable task) {
    long enqueued = System.nanoTime();
    submitted.increment();
    pool.execute(() -> {
        queueWaitNanos.add(System.nanoTime() - enqueued);
        try { task.run(); } finally { completed.increment(); }
    });
}

public long averageQueueWaitMillis() {
    long done = completed.sum();
    return done == 0L ? 0L : TimeUnit.NANOSECONDS.toMillis(queueWaitNanos.sum() / done);
}

enqueued 在 pool.execute 之前取值，被拒绝的任务不会进入这个统计；如果要连拒绝一起看，用 submitted 和 completed + rejected 的差值对账。

调用点：从业务线程回到事件循环

业务线程只拿到不可变的请求数据和一份响应编码结果，不拿 Connection。写回必须经过 loop.execute：

try {
    business.execute(() -> {
        byte[] encoded = encode(route(requestCopy));
        loop.execute(() -> connection.enqueue(ByteBuffer.wrap(encoded)));
    });
} catch (RejectedExecutionException e) {
    byte[] encoded = encode(serviceUnavailable());
    loop.execute(() -> connection.enqueue(ByteBuffer.wrap(encoded)));
}

这段代码有四个容易漏的约束：

1. requestCopy 必须在提交前复制。 HttpParser 可能复用读缓冲区，HttpRequest.body() 是 byte[]，直接把这个数组交给业务线程，下一次读会覆盖它。复制 body 和 headers，再构造新的 HttpRequest。
2. 响应在业务线程编码完。 encode 可能做字符集转换、压缩、序列化，都属于业务工作；不要在循环线程上做。
3. ByteBuffer 的所有权一次转移。 ByteBuffer.wrap(encoded) 创建的是堆缓冲区，交给 Connection.enqueue 后由写队列负责释放；如果换成池化缓冲区，释放路径必须和 Connection.close() 对齐。
4. 拒绝路径也要回事件循环。 503 的编码可以在调用线程做（很短），但 connection.enqueue 仍然要 loop.execute；调用线程可能是事件循环本身（比如 accept 后的快速拒绝），也可能不是。

不要在业务线程里 future.get() 等另一个业务任务，也不要在事件循环里等业务结果。需要编排多个下游时，用回调或 CompletableFuture 的 thenAccept，最后一步仍然是 loop.execute。

任务进池前还要看截止时间。RequestContext.deadline（12.3）给出本次请求还剩多少预算；已经过期的请求直接回 504，不要再占用业务线程去访问下游，否则下游超时、重试和队列排队会叠加成雪崩。任务开始和写回前各检查一次，剩余预算小于下游 p99 时降级，不要抱着「再试一次」的心态把线程占住。

常见错误与边界

错误一：用 Executors.newFixedThreadPool()。 它内部是 LinkedBlockingQueue，默认容量 Integer.MAX_VALUE；下游一慢，队列涨到 OOM 才停。必须显式 new ThreadPoolExecutor(...) 并传有界队列。

错误二：配 CallerRunsPolicy。 池子满时由提交者执行；提交者是事件循环，就等于在循环上跑业务。低并发压测触发不到，线上高负载时突然出现心跳延迟。linkd 用「计数 + 抛异常 + 回 503」。

错误三：业务线程直接 connection.enqueue(...)。 outbound 是 ArrayDeque，interestOps 也不是线程安全的；轻则丢响应，重则 selector 状态错乱。所有写回都经过 loop.execute。

错误四：把解析缓冲区直接传给业务线程。 解析器复用 ByteBuffer 或 byte[]，业务线程读的时候可能已经被下一次读覆盖。跨线程前复制请求数据。

错误五：只调 shutdown() 不等。 在途任务还在跑，主线程已经退出，响应写不回去；或者 shutdownNow() 中断了正在写半条响应的任务，客户端收到截断的帧。关闭要分阶段，并配合连接排空。

错误六：用虚拟线程但不限并发。 每请求一个虚拟线程，下游连接池只有 50 个连接，虚拟线程会成百上千地堆在连接池的等待队列上，超时和重试一起放大。用 Semaphore 或下游连接池做并发上限。

边界：线程数不是越大越好。 线程多了，上下文切换、栈内存、下游连接数都会涨；等待比公式给的是起点，最终以压测的 p99 和拒绝率为准。10 核机器上开 500 条业务线程，通常只是把排队从队列挪到了 CPU 调度器。

边界：业务池不解决 CPU 密集。 纯计算任务放在业务池里，只会占满线程；要不另开 CPU 池，要不把任务拆小、限制并发。虚拟线程对 CPU 密集也没有帮助。

小结

• 业务线程池的第一目标是隔离阻塞，不是提高吞吐；线程数按等待比估算，再用压测校正。
• 队列必须有界，拒绝策略必须显式；CallerRunsPolicy 会把阻塞带回事件循环，linkd 禁止使用。
• 拒绝是背压信号：计数、抛 RejectedExecutionException、回 503，比无界排队更可控。
• 业务线程只碰不可变请求数据和编码后的响应，写回统一 loop.execute；请求缓冲区跨线程前必须复制。
• 虚拟线程适合高并发阻塞型业务，不替代有界并发，也不加速 CPU 密集任务；下游容量要用信号量或连接池兜住。
• BusinessExecutor 用 LongAdder 暴露提交、完成、拒绝、队列深度和活跃线程数；拒绝率和排队时间比瞬时队列长度更值得报警。

线程池解决了「在哪里跑」，但请求的身份、租户、截止时间还要跟着任务走。下一节看 ThreadLocal 为什么会把上一个请求的用户带给下一个请求，以及 ScopedValue 在事件循环和业务线程之间应该怎么绑定、怎么恢复。

阅读导航：上一节：12.1 事件循环与任务队列 · 下一节：12.3 上下文传播与线程安全 。