java 线程池面试要点

// 不用线程池：每次新建，开销大
new Thread(() -> doWork()).start();

// 用线程池：复用线程
ExecutorService pool = Executors.newFixedThreadPool(10);
pool.execute(() -> doWork());

public ThreadPoolExecutor(
    int corePoolSize,                  // 1. 核心线程数
    int maximumPoolSize,               // 2. 最大线程数
    long keepAliveTime,                // 3. 空闲存活时间
    TimeUnit unit,                     // 4. 时间单位
    BlockingQueue<Runnable> workQueue, // 5. 任务队列
    ThreadFactory threadFactory,       // 6. 线程工厂
    RejectedExecutionHandler handler   // 7. 拒绝策略
)

corePoolSize ≤ 当前线程数 ≤ maximumPoolSize

// 非核心线程空闲 60 秒后回收
new ThreadPoolExecutor(2, 10, 60L, TimeUnit.SECONDS, ...);

// 有界队列，推荐
new ArrayBlockingQueue<>(100);

// 无界队列，有 OOM 风险
new LinkedBlockingQueue<>();

ThreadFactory factory = new ThreadFactory() {
    private final AtomicInteger count = new AtomicInteger(1);
    @Override
    public Thread newThread(Runnable r) {
        Thread t = new Thread(r, "my-pool-" + count.getAndIncrement());
        t.setDaemon(false);
        t.setUncaughtExceptionHandler((thread, e) -> 
            log.error("线程异常", e));
        return t;
    }
};

提交任务
   ↓
1. 核心线程未满？ → 创建核心线程执行
   ↓ 否
2. 队列未满？ → 放入队列等待
   ↓ 否
3. 线程数 < max？ → 创建非核心线程执行
   ↓ 否
4. 执行拒绝策略

execute(task)
    │
    ├─ 线程数 < corePoolSize → addWorker(core)
    │
    ├─ 队列未满 → workQueue.offer(task)
    │
    ├─ 线程数 < maximumPoolSize → addWorker(non-core)
    │
    └─ 拒绝策略

ThreadPoolExecutor pool = new ThreadPoolExecutor(...);
pool.prestartAllCoreThreads();  // 启动所有核心线程
pool.prestartCoreThread();      // 启动一个核心线程

final void runWorker(Worker w) {
    Runnable task = w.firstTask;
    w.firstTask = null;
    while (task != null || (task = getTask()) != null) {
        task.run();  // 执行任务
        task = null;
    }
}

// getTask 中的回收逻辑
if ((wc > maximumPoolSize || (timed && timedOut))
    && (wc > 1 || workQueue.isEmpty())) {
    if (compareAndDecrementWorkerCount(c))
        return null;  // 返回 null，线程退出
}

// execute：异常直接抛出
pool.execute(() -> { throw new RuntimeException(); });

// submit：异常在 get() 时抛出
Future<?> f = pool.submit(() -> { throw new RuntimeException(); });
f.get(); // 抛 ExecutionException

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

// 高 3 位：线程池状态
// 低 29 位：工作线程数
private static final int COUNT_BITS = Integer.SIZE - 3; // 29
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

// 状态
RUNNING    = -1 << 29  // 接受新任务，处理队列任务
SHUTDOWN   =  0 << 29  // 不接受新任务，处理队列任务
STOP       =  1 << 29  // 不接受新任务，不处理队列，中断线程
TIDYING    =  2 << 29  // 所有任务终止，即将调用 terminated()
TERMINATED =  3 << 29  // terminated() 完成

RUNNING ──shutdown()──→ SHUTDOWN ──队列空+线程空──→ TIDYING ──→ TERMINATED
   │                                                    
   └──shutdownNow()──→ STOP ──线程空──→ TIDYING ──→ TERMINATED

pool.shutdown();     // 优雅关闭
pool.shutdownNow();  // 强制关闭，返回未执行任务

pool.shutdown();
if (!pool.awaitTermination(60, TimeUnit.SECONDS)) {
    pool.shutdownNow();
}

pool.shutdown();
System.out.println(pool.isShutdown());   // true
System.out.println(pool.isTerminated()); // false，可能还有任务
// 等待所有任务完成
pool.awaitTermination(1, TimeUnit.MINUTES);
System.out.println(pool.isTerminated()); // true

// 触发拒绝的场景
ThreadPoolExecutor pool = new ThreadPoolExecutor(
    1, 1, 0, TimeUnit.SECONDS,
    new ArrayBlockingQueue<>(1)); // 队列容量 1

pool.execute(task1); // 核心线程执行
pool.execute(task2); // 入队列
pool.execute(task3); // 触发拒绝策略

// 1. AbortPolicy（默认）
new ThreadPoolExecutor.AbortPolicy();

// 2. CallerRunsPolicy
new ThreadPoolExecutor.CallerRunsPolicy();

// 3. DiscardPolicy
new ThreadPoolExecutor.DiscardPolicy();

// 4. DiscardOldestPolicy
new ThreadPoolExecutor.DiscardOldestPolicy();

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    if (!e.isShutdown()) {
        r.run();  // 调用者线程执行
    }
}

public class CustomRejectPolicy implements RejectedExecutionHandler {
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        // 1. 记录日志
        log.warn("任务被拒绝: {}", r);
        
        // 2. 持久化到数据库/MQ，后续补偿
        saveToDb(r);
        
        // 3. 或降级处理
        // fallback(r);
    }
}

int cpuCores = Runtime.getRuntime().availableProcessors();

// CPU 密集型
int cpuPoolSize = cpuCores + 1;

// IO 密集型
int ioPoolSize = cpuCores * 2;

线程数 = N × (1 + 等待时间 / 计算时间)

// 例：CPU 4 核，等待 100ms，计算 50ms
// 线程数 = 4 × (1 + 100/50) = 12

ThreadPoolExecutor pool = new ThreadPoolExecutor(...);
pool.setCorePoolSize(20);
pool.setMaximumPoolSize(50);
pool.setKeepAliveTime(30, TimeUnit.SECONDS);

pool.getPoolSize();          // 当前线程数
pool.getCorePoolSize();      // 核心线程数
pool.getMaximumPoolSize();   // 最大线程数
pool.getActiveCount();       // 活跃线程数
pool.getQueue().size();      // 队列任务数
pool.getCompletedTaskCount();// 已完成任务数
pool.getTaskCount();         // 总任务数
pool.getLargestPoolSize();   // 历史最大线程数

// 危险：无界队列
public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
        0L, TimeUnit.MILLISECONDS,
        new LinkedBlockingQueue<Runnable>()); // 无界！
}

ThreadPoolExecutor pool = new ThreadPoolExecutor(
    8,                              // core：CPU 核数
    16,                             // max：2 倍
    60L, TimeUnit.SECONDS,          // 空闲回收
    new ArrayBlockingQueue<>(200),  // 有界队列
    new ThreadFactoryBuilder()
        .setNameFormat("biz-pool-%d")
        .build(),
    new ThreadPoolExecutor.CallerRunsPolicy() // 不丢任务
);

// 允许核心线程超时回收
pool.allowCoreThreadTimeOut(true);

// 预启动核心线程
pool.prestartAllCoreThreads();

private boolean addWorker(Runnable firstTask, boolean core) {
    // 1. CAS 增加 workerCount
    // 2. new Worker(firstTask)
    // 3. 启动线程 t.start()
    // 4. 加入 workers 集合
}

private Runnable getTask() {
    boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
    for (;;) {
        if (timed) {
            // 非核心线程：超时获取
            Runnable r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS);
            if (r != null) return r;
            timedOut = true;
        } else {
            // 核心线程：阻塞获取
            Runnable r = workQueue.take();
            if (r != null) return r;
        }
        // 超时或状态异常，返回 null，线程退出
    }
}

private final class Worker extends AbstractQueuedSynchronizer 
        implements Runnable {
    final Thread thread;
    Runnable firstTask;
    volatile long completedTasks;
}

pool.execute(() -> {
    throw new RuntimeException("error");
});
// 线程打印异常后死亡，线程池补充新线程

Future<?> f = pool.submit(() -> {
    throw new RuntimeException("error");
});
try {
    f.get(); // 抛 ExecutionException
} catch (ExecutionException e) {
    Throwable cause = e.getCause(); // 原始异常
}

// processWorkerExit 简化
if (runState < STOP) {
    addWorker(null, false); // 补充线程
}

public void shutdownGracefully(ExecutorService pool) {
    pool.shutdown(); // 1. 拒绝新任务，处理队列
    try {
        if (!pool.awaitTermination(60, TimeUnit.SECONDS)) {
            pool.shutdownNow(); // 2. 超时强制关闭
            if (!pool.awaitTermination(60, TimeUnit.SECONDS)) {
                System.err.println("线程池未终止");
            }
        }
    } catch (InterruptedException e) {
        pool.shutdownNow(); // 3. 当前线程中断，也强制关闭
        Thread.currentThread().interrupt(); // 恢复中断
    }
}

new ThreadPoolExecutor(0, Integer.MAX_VALUE,
    60L, TimeUnit.SECONDS,
    new SynchronousQueue<>()); // CachedThreadPool

线程A: [任务1, 任务2, 任务3]  ← 从头部取
线程B: [任务4, 任务5]         ← 空闲，从A尾部偷任务3

core = 20, max = 50, queue = 500
// 请求多，IO 等待长

core = N+1, max = N+1, queue = 100
// 计算为主，线程不宜多

core = 10, max = 10, queue = 1000
// 稳定吞吐，队列缓冲

Future<?> f = pool.submit(task);
try {
    f.get(3, TimeUnit.SECONDS);
} catch (TimeoutException e) {
    f.cancel(true); // 中断任务
}

pool.execute(() -> {
    long start = System.currentTimeMillis();
    while (!Thread.interrupted()) {
        if (System.currentTimeMillis() - start > 3000) break;
        // 工作...
    }
});

// 危险：无界队列
new LinkedBlockingQueue<>();

// 安全：有界队列
new ArrayBlockingQueue<>(1000);

jstack <pid> | grep "pool-"

thread --state WAITING
thread -n 3  # 最忙的3个线程

scheduledExecutor.scheduleAtFixedRate(() -> {
    log.info("pool: active={}, queue={}, completed={}",
        pool.getActiveCount(),
        pool.getQueue().size(),
        pool.getCompletedTaskCount());
}, 0, 10, TimeUnit.SECONDS);

// 不好的命名
Thread-0, Thread-1, pool-1-thread-1

// 好的命名
order-pool-1, payment-pool-2, sms-pool-1