Java · 线程池 ThreadPoolExecutor

场景 1 · 标准的生产级线程池

ThreadPoolExecutor pool = new ThreadPoolExecutor(
    8, 16,                                     // core / max: 压测起点
    60, TimeUnit.SECONDS,
    new ArrayBlockingQueue<>(500),                  // 有界! 水位=缓冲突发, 不是垃圾桶
    new ThreadFactoryBuilder()
        .setNameFormat("order-export-%d").build(),     // 命名: jstack 一眼定位
    new ThreadPoolExecutor.CallerRunsPolicy());       // 反压 (或自定义降级)

场景 2 · 监控水位 + 动态调参

// 暴露到 Prometheus:
gauge.set(pool.getActiveCount(), pool.getQueue().size(), rejectedCount.get())
// 告警联动配置中心 (Apollo/Nacos) 热调参, 不重启:
pool.setCorePoolSize(newCore);      // 运行时生效
pool.setMaximumPoolSize(newMax);
// 大促前: 队列水位 >80% → 自动 core 8→32, 平峰回收

场景 3 · submit 吞异常排查

Future<?> f = pool.submit(task);
try { f.get(3, TimeUnit.SECONDS); }          // 不 get: 任务死了都没声
catch (ExecutionException e) {
    log.error("task failed", e.getCause());   // 真实业务异常在 cause 里
}

场景 4 · Web 容器与业务池隔离

Tomcat 线程只做接入, 慢业务(导出/推送)必须独立池, 防互相拖垮:

// Tomcat: 快速接入 (server.tomcat.threads.max=200)
# 业务: 慢操作独立池 — 导出任务再慢也不影响下单接口
@Bean("exportPool")
public ThreadPoolExecutor exportPool() {
    return new ThreadPoolExecutor(4, 8, 60, TimeUnit.SECONDS,
        new ArrayBlockingQueue<>(200),
        namedThreadFactory("export"),      // 命名区分: jstack 一眼归属
        new ThreadPoolExecutor.AbortPolicy());
}
# @Async("exportPool") 或手动 submit — 明确指定, 不用默认池

场景 5 · CompletionService: 谁先完成先处理

并行调 N 个下游, 按完成顺序消费而不是提交顺序等待:

ExecutorCompletionService<Quote> ecs =
        new ExecutorCompletionService<>(pool);
for (Vendor v : vendors) ecs.submit(() -> v.quote(req));

for (int i = 0; i < vendors.size(); i++) {
    Future<Quote> f = ecs.poll(500, MILLISECONDS);    // 谁先回先拿
    if (f != null) merge(f.get());               // 超时不候: 慢 vendor 不拖整体
}

场景 6 · CompletableFuture 编排 + 自定义池

声明式并发编排, 但默认池是 ForkJoinPool.commonPool — 生产必须显式传池:

CompletableFuture
    .supplyAsync(() -> queryUser(uid), ioPool)       // 显式业务池!
    .thenCombine(
        CompletableFuture.supplyAsync(() -> queryOrders(uid), ioPool),
        (user, orders) -> render(user, orders))
    .orTimeout(800, MILLISECONDS)                 // JDK9+ 整链超时
    .exceptionally(e -> fallbackPage());                 // 降级页兜底

场景 7 · 拒绝时落 MQ 延迟重试

自定义拒绝策略: 不丢任务, 转投消息队列削峰:

new ThreadPoolExecutor.RejectedExecutionHandler() {
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        log.warn("pool saturated, queue={}", e.getQueue().size());
        rejectedCounter.inc();                          // 拒绝计数: 核心告警指标
        mqProducer.send("task-retry", serialize(r), delay(30s));  // 30s 后回放
    }
}

场景 8 · 舱壁隔离: 核心接口独立池

支付回调与营销推送共用池 = 推送风暴时支付被拖死; 按重要性分组隔离:

pools.put("payment", newPool(32, 32, 1000));   // 核心交易: 大配额
pools.put("marketing", newPool(4, 8, 200));    // 可降级: 小配额+丢弃策略
# 每个池独立监控水位与拒绝数 — 一个池打满, 告警只指向那个业务域

场景 9 · 压测找容量拐点

线程数不是猜的 — 固定负载扫描, 找吞吐拐点与队列水位的关系:

// 阶梯压测: 逐步加压, 每档记录三件事
for threads in 50 100 150 200 300; do
    run_load --threads=$threads --duration=120s
    record: QPS | P99 | pool.active + queue.size   # 三线同图
done
# 拐点特征: QPS 不再涨而 queue 持续涨 → 该档位就是容量上限, 参数定在拐点前 20%

场景 10 · ScheduledThreadPoolExecutor 的正确用法

定时任务池的两个必知: 异常会杀掉后续调度、任务要幂等:

ScheduledExecutorService ses = Executors.newScheduledThreadPool(4,
        namedThreadFactory("sync"));
ses.scheduleWithFixedDelay(() -> {
    try { syncFromUpstream(); }               // 必须 try 包住!
    catch (Throwable t) { log.error("sync fail", t); }  // 抛出=该任务永久停摆
}, 0, 30, TimeUnit.SECONDS);                // fixedDelay: 上一轮跑完再计时