books

《Java 并发编程·线程与内存模型》10.3 监控、动态调参与关闭

线程池上线之后要回答三个运维问题：指标怎么读、参数能不能改、进程怎么退。本节逐个核对 getActiveCount、队列长度、getCompletedTaskCount 的口径与误差，实测 setCorePoolSize 上调立即补线程、下调却要等 keepAlive 才收敛，并对比 shutdown 与 shutdownNow 的返回值和中断行为。

Leeting Yan 10 分钟阅读 4590 字

承接的问题

10.1 讲怎么配，10.2 讲配满了怎么办。这一节讲上线之后：池子跑起来，你怎么知道它健康；流量变了，参数能不能不重启就改；发布或下线时，怎么让队列里的任务体面地做完。

三个问题对应三组 API：getActiveCount()、getQueue().size()、getCompletedTaskCount() 读到的到底是什么；setCorePoolSize() 调下去是立刻生效还是慢慢收敛；shutdown() 和 shutdownNow() 差在哪、两段式关闭为什么是标准写法。它们的共同点是：API 的语义比名字复杂。getActiveCount() 听起来像「正在干活的任务数」，实际是一个会抖动的估计值；setCorePoolSize() 听起来像「立刻改成 N 个线程」，实际只有上调是立刻的。

指标的口径与误差

ThreadPoolExecutor 暴露了一组 getter，全部可以直接读，不需要 JMX：

方法含义口径误差来源
getPoolSize()当前线程数工作集合大小线程退出与移除之间有窗口
getCorePoolSize() / getMaximumPoolSize()配置值精确无
getActiveCount()正在执行任务的线程数近似遍历 worker 时逐个判断是否持锁，遍历期间状态会变
getQueue().size()排队任务数队列自身精确不含已交给线程、尚未开始执行的任务
getTaskCount()历史提交总数近似各 worker 计数之和，非原子快照
getCompletedTaskCount()已完成总数近似同上，完成与读取之间有窗口
getLargestPoolSize()历史线程数峰值精确（单调）无

Javadoc 对 getActiveCount() 的原话是「approximate number of threads that are actively executing tasks」，对两个计数器的原话也是「approximate」。这不是实现偷懒，而是这些值要跨多个 worker 汇总，不可能在不暂停池的情况下取到一致快照。

实测：一次快照能读出什么

MonitorProbe 建了一个 core=2, max=4, queue=10 的池，提交 6 个阻塞任务，等状态稳定后打印一次快照，然后释放、关闭、再打印一次：

import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.BooleanSupplier;

public final class MonitorProbe {

    public static void main(String[] args) throws Exception {
        snapshot();
        resize();
        shutdownVsNow();
    }

    static void printState(String tag, ThreadPoolExecutor pool) {
        System.out.println(tag
                + " core=" + pool.getCorePoolSize()
                + " pool=" + pool.getPoolSize()
                + " active=" + pool.getActiveCount()
                + " queue=" + pool.getQueue().size()
                + " taskCount=" + pool.getTaskCount()
                + " completed=" + pool.getCompletedTaskCount()
                + " largest=" + pool.getLargestPoolSize());
    }

    static void snapshot() throws Exception {
        ThreadPoolExecutor pool = new ThreadPoolExecutor(
                2, 4, 30, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(10), r -> new Thread(r, "loan-mon"));
        CountDownLatch release = new CountDownLatch(1);
        for (int i = 0; i < 6; i++) {
            pool.execute(() -> {
                try {
                    release.await();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
        }
        waitUntil(() -> pool.getActiveCount() == 2 && pool.getQueue().size() == 4, 5_000);
        printState("running", pool);

        release.countDown();
        pool.shutdown();
        boolean done = pool.awaitTermination(10, TimeUnit.SECONDS);
        printState("after  ", pool);
        System.out.println("terminated=" + done);
    }

    static void resize() throws Exception {
        ThreadPoolExecutor pool = new ThreadPoolExecutor(
                1, 8, 2, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(), r -> new Thread(r, "loan-resize"));
        CountDownLatch release = new CountDownLatch(1);
        for (int i = 0; i < 5; i++) {
            pool.execute(() -> {
                try {
                    release.await();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
        }
        waitUntil(() -> pool.getActiveCount() == 1 && pool.getQueue().size() == 4, 5_000);
        printState("beforeResize", pool);

        pool.setCorePoolSize(4);
        boolean grew = waitUntil(() -> pool.getActiveCount() == 4, 5_000);
        printState("afterGrow  ", pool);
        System.out.println("observedGrow=" + grew);

        release.countDown();
        waitUntil(() -> pool.getCompletedTaskCount() == 5, 10_000);
        pool.setCorePoolSize(1);
        boolean shrank = waitUntil(() -> pool.getPoolSize() == 1, 20_000);
        printState("afterShrink", pool);
        System.out.println("observedShrink=" + shrank);
        pool.shutdown();
        pool.awaitTermination(5, TimeUnit.SECONDS);
    }

    static void shutdownVsNow() throws Exception {
        for (String mode : List.of("shutdown", "shutdownNow")) {
            ThreadPoolExecutor pool = new ThreadPoolExecutor(
                    1, 1, 0, TimeUnit.SECONDS,
                    new ArrayBlockingQueue<>(5), r -> new Thread(r, "loan-close"));
            AtomicBoolean interrupted = new AtomicBoolean();
            CountDownLatch started = new CountDownLatch(1);
            pool.execute(() -> {
                started.countDown();
                try {
                    Thread.sleep(1_000);
                } catch (InterruptedException e) {
                    interrupted.set(true);
                }
            });
            started.await();
            pool.execute(() -> { });
            pool.execute(() -> { });

            int pending = 0;
            if (mode.equals("shutdown")) {
                pool.shutdown();
            } else {
                List<Runnable> never = pool.shutdownNow();
                pending = never.size();
            }
            boolean terminated = pool.awaitTermination(5, TimeUnit.SECONDS);
            System.out.println(mode + " pending=" + pending
                    + " interrupted=" + interrupted.get()
                    + " completed=" + pool.getCompletedTaskCount()
                    + " terminated=" + terminated);
        }
    }

    static boolean waitUntil(BooleanSupplier cond, long timeoutMs) throws InterruptedException {
        long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(timeoutMs);
        while (System.nanoTime() < deadline) {
            if (cond.getAsBoolean()) {
                return true;
            }
            Thread.sleep(5);
        }
        return cond.getAsBoolean();
    }
}

编译运行：

export JAVA_HOME=~/.jdks/jdk-25.0.5+8/Contents/Home
$JAVA_HOME/bin/javac --release 25 -d classes MonitorProbe.java
$JAVA_HOME/bin/java -cp classes MonitorProbe

本机输出：

running core=2 pool=2 active=2 queue=4 taskCount=6 completed=0 largest=2
after   core=2 pool=0 active=0 queue=0 taskCount=6 completed=6 largest=2
terminated=true
beforeResize core=1 pool=1 active=1 queue=4 taskCount=5 completed=0 largest=1
afterGrow   core=4 pool=4 active=4 queue=1 taskCount=5 completed=0 largest=4
observedGrow=true
afterShrink core=1 pool=1 active=0 queue=0 taskCount=5 completed=5 largest=4
observedShrink=true
shutdown pending=0 interrupted=false completed=3 terminated=true
shutdownNow pending=2 interrupted=true completed=1 terminated=true

running 那一行是最有用的一次快照：taskCount=6、completed=0、queue=4、active=2。三者满足 6 - 0 = 4 + 2——总提交 = 已完成 + 排队中 + 正在执行。这个恒等式在任意时刻近似成立，可以用来交叉校验指标有没有读错：如果 queue + active 明显大于 taskCount - completed，说明读到的不是同一时刻的状态。

after 那一行 pool=0 是关闭完成后的样子：worker 全部退出，线程池对象还在，但已经没有任何线程。此时 getCompletedTaskCount() 仍是 6——关闭不会清空计数器，这对排查「停机前到底跑完多少」很有用。三个使用建议：

  • 不要用 getActiveCount() 做限流。它是估计值，且统计的是「持有 worker 锁的线程」，不区分任务是 CPU 密集还是阻塞。要限流用 Semaphore（14.3 节）。
  • 队列长度要盯趋势，不要盯瞬时值。单次 queue.size() 没有意义，持续 5 分钟上涨才是信号。
  • 计数器只用于监控和展示，不要用于计费、对账这类需要精确性的逻辑。

采样线程

指标要周期采集，用池外一个 daemon 平台线程即可：循环里读四个 getter、打印、Thread.sleep，pool.isShutdown() 为真时退出。不要用被监控的这个池去跑监控任务——池满时监控会跟着停摆；也不需要用 ScheduledThreadPoolExecutor（11.1 节），监控线程没有定时精度要求。daemon 属性保证它不阻止 JVM 退出。

setCorePoolSize 的即时影响

resize() 的实验：core=1, max=8, keepAlive=2s，无界队列，先提交 5 个阻塞任务（1 个在跑、4 个排队），然后 setCorePoolSize(4)：

beforeResize core=1 pool=1 active=1 queue=4 taskCount=5 completed=0 largest=1
afterGrow   core=4 pool=4 active=4 queue=1 taskCount=5 completed=0 largest=4

上调是立刻生效的：线程数从 1 涨到 4，队列从 4 降到 1。机制是 setCorePoolSize 发现新值大于旧值，就按 min(delta, queue.size()) 补建 worker，新 worker 直接从队列里取任务。这里是 delta=3、队列 4 个，所以补 3 个，剩 1 个继续排队。

下调则不是立刻的：

afterShrink core=1 pool=1 active=0 queue=0 taskCount=5 completed=5 largest=4
observedShrink=true

这一行看起来「立刻生效」，但它是在 waitUntil(poolSize == 1, 20s) 成立之后才打印的。把 keepAlive 从 2 秒改成 30 秒、其余不变再跑一次，输出是：

grow core=4 pool=4
keepAlive=30s after 1s  pool=4 core=1
keepAlive=30s after 10s pool=4 core=1

core 已经是 1，pool 却还是 4。原因在 setCorePoolSize 的下调分支：它只调用 interruptIdleWorkers() 唤醒空闲线程，不销毁线程；线程被唤醒后重新检查退出条件，发现自己超出 core，于是用 keepAlive 作为超时继续 poll，超时才退出。所以：

  • 上调：立即补线程，最多补 delta 个，且不超过队列长度；
  • 下调：只是「不再续命」，真正收敛要等 keepAliveTime；keepAlive 30 秒的池，缩容也要 30 秒起步。

这条差异直接决定运维动作的含义：扩容量是止血，缩容量是慢撒气。大促前把 core 调大能立刻生效；大促后调回去，别指望线程数马上降，也别拿「poolSize 没降」当成调参失败。

两个参数校验在本机 JDK 25 实测过：

  • setCorePoolSize(10) 在 max=4 时抛 IllegalArgumentException: corePoolSize must be less than or equal to maximumPoolSize；
  • setMaximumPoolSize(1) 在 core=2 时抛 maximumPoolSize must be greater than or equal to corePoolSize。

所以动态调参有顺序要求：扩容先调 max 再调 core，缩容先调 core 再调 max。反过来会撞上上面两个异常。另外 setKeepAliveTime(0, SECONDS) 在未开启核心超时时是允许的，只有开启 allowCoreThreadTimeOut 后才要求非零。

动态调参还有两个边界：队列容量改不了（ArrayBlockingQueue 构造时定死），下游容量也改不了（core 调到 64，数据库连接池还是 20，多出来的线程只会排队等连接）。后者是「调参调出事故」的常见原因。

shutdown 与 shutdownNow 两段式

两个方法的差别不在名字，在返回值和对运行中任务的处理：

维度shutdown()shutdownNow()
接受新任务否否
队列中的任务继续执行完不再执行，作为返回值返回
正在执行的任务不受影响尝试 interrupt
返回值voidList<Runnable>（从未开始的任务）
是否保证停止是（只要有任务最终会结束）否，只是「尽力」

实测（1 个运行中的任务睡 1 秒，队列里 2 个空任务）：

shutdown pending=0 interrupted=false completed=3 terminated=true
shutdownNow pending=2 interrupted=true completed=1 terminated=true
  • shutdown：pending=0（没有任务被退回），运行中的任务没被中断，最终 completed=3——队列里的 2 个也执行了。
  • shutdownNow：pending=2（2 个排队任务被退回），运行中的任务被中断（interrupted=true），最终 completed=1。注意它仍然 completed=1：任务捕获了 InterruptedException 后正常返回，所以算「完成」；如果任务把中断吞掉继续跑，completed 会一直不涨。

shutdownNow 最容易被误解的地方，是它并不保证线程停下来。另一个最小实验（StubbornProbe，1 个线程、队列 5 个）让任务在 catch (InterruptedException) 里什么都不做、继续循环 3 秒，然后 shutdownNow 并等 1 秒：

stubborn terminatedAfter1s=false poolSize=1

任务吞掉了中断标志，继续占着那唯一的线程。shutdownNow 发了中断，但中断只是一个标志位加一次 InterruptedException，任务不响应就没有任何强制手段。awaitTermination(1s) 返回 false，线程池还活着。shutdownNow 不是 kill -9，它依赖任务本身可中断（第 6 章）。

标准的两段式关闭：

public void close() throws InterruptedException {
    pool.shutdown();                                   // 第一段：不再收新任务
    if (!pool.awaitTermination(5, TimeUnit.SECONDS)) { // 等一个上限
        List<Runnable> never = pool.shutdownNow();     // 第二段：中断 + 退回队列
        System.out.println("dropped " + never.size() + " queued tasks");
        if (!pool.awaitTermination(2, TimeUnit.SECONDS)) {
            System.err.println("pool still alive after shutdownNow");
        }
    }
}

三个细节：

  • awaitTermination 本身不会中断任何线程，它只是等；等待期间线程被中断会抛 InterruptedException，调用方要恢复标志。
  • 第二段之后要再等一次，确认池真的停了；返回 false 说明有任务不响应中断，此时只能记录告警，不能假装已经关干净。
  • shutdownNow 返回的 List<Runnable> 里是从未开始执行的任务；已经跑了一半的任务不在里面。要保证不丢，得在任务层面做幂等和补偿（卷三展开）。

关闭和 JVM 退出的配合（addShutdownHook）在第 6.3 节讲过，要点相同：hook 里不能做长耗时操作，关闭顺序不确定，异常会被吞掉。

在 libraryd-conc 里落地

PooledLoanService 在 10.1、10.2 的基础上补两个能力：一个只读的快照方法，供监控线程和诊断接口调用；一个两段式 close()，不依赖容器的关闭钩子也能安全停机。

// 字段与构造器同 10.2，这里只给新增方法
public java.util.Map<String, Long> snapshot() {
    java.util.Map<String, Long> m = new java.util.LinkedHashMap<>();
    m.put("core", (long) pool.getCorePoolSize());
    m.put("max", (long) pool.getMaximumPoolSize());
    m.put("pool", (long) pool.getPoolSize());
    m.put("active", (long) pool.getActiveCount());
    m.put("queue", (long) pool.getQueue().size());
    m.put("largest", (long) pool.getLargestPoolSize());
    m.put("taskCount", pool.getTaskCount());
    m.put("completed", pool.getCompletedTaskCount());
    m.put("rejected", rejection.rejectedCount());
    return m;
}

@Override
public void close() throws InterruptedException {
    pool.shutdown();
    if (!pool.awaitTermination(5, TimeUnit.SECONDS)) {
        java.util.List<Runnable> never = pool.shutdownNow();
        System.err.println("loan-pool dropped " + never.size() + " queued tasks");
        if (!pool.awaitTermination(2, TimeUnit.SECONDS)) {
            System.err.println("loan-pool still alive after shutdownNow");
        }
    }
}

snapshot() 里没有加同步，也没有停顿线程池：这些 getter 本身就是近似值，加锁只会制造「看起来精确」的假象。真正需要一致性的场景（比如计费）不应该读线程池指标，而应该在任务完成时自己累加计数器。

close() 的 5 秒和 2 秒是两层预算：第一层留给正常任务跑完，第二层留给中断生效。两个数字都应该小于发布系统的终止宽限期（Kubernetes 默认 30 秒），否则进程会先被 SIGKILL，队列里的任务还是丢。

常见错误与边界

错误后果改法
用 getActiveCount() 判断是否该扩容估计值抖动，扩容决策跟着抖看队列长度趋势和任务延迟
把 getCompletedTaskCount() 当精确值对账完成瞬间与读取瞬间有窗口，数字对不上业务侧自己累加
只调 core 不调 max上调时抛 IllegalArgumentException扩容先 max 后 core，缩容反之
调大 core 却不管下游线程变多，连接池没变，全部排队等连接同步评估下游容量
只调 shutdown() 不 awaitTermination进程退出，队列任务丢失两段式 + 两层超时
以为 shutdownNow 一定能停不响应中断的任务继续跑任务内响应中断 + 幂等
在关闭流程里忽略返回值不知道有多少任务被退回记录 never.size() 并告警
监控线程用被监控的池池满时监控也停摆独立守护线程或独立池

什么时候不要做动态调参：当参数变化需要同时改队列容量、连接池大小、超时阈值时，单改 core 只会制造新的不均衡。这类变更应该走配置发布 + 重启，而不是运行期热改。热改适合的是「临时扩容止血」这一种场景。

小结

  • getActiveCount()、getTaskCount()、getCompletedTaskCount() 都是近似值，getQueue().size() 不含已交给线程的任务；taskCount - completed ≈ queue + active 可以交叉校验。
  • 本机快照 taskCount=6 completed=0 queue=4 active=2 满足该恒等式；关闭后计数器保留，completed=6。
  • setCorePoolSize 上调立刻补线程（实测 pool 1→4、queue 4→1）；下调只唤醒空闲线程，真正退出要等 keepAlive（keepAlive=30s 时 10 秒后仍是 4 个线程）。
  • 参数校验要求 core ≤ max，所以扩容先改 max、缩容先改 core；队列容量运行期不可改。
  • shutdown 让队列任务跑完，shutdownNow 退回 2 个排队任务并中断运行中的任务；不响应中断的任务会让 awaitTermination 返回 false。
  • 两段式关闭 = shutdown + awaitTermination + shutdownNow + 再等一次，两层超时都要小于进程终止宽限期。

下一章进入定时任务：Timer 为什么会被一个异常整个干掉，ScheduledThreadPoolExecutor 又是怎么在同样的异常下继续工作的。

阅读导航：上一节 10.2 队列与拒绝策略 ｜ 下一节 11.1 Timer 的缺陷与 ScheduledThreadPoolExecutor

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「books」更多文章

  1. 《Java 消息中间件·事件驱动架构》附录 B 架构决策记录模板
  2. 《Java 消息中间件·事件驱动架构》附录 A 术语表
  3. 《Java 消息中间件·事件驱动架构》15.3 全书验收