多线程与操作系统相关手搓线程池

我不只是告诉你代码是什么，更会告诉你为什么这么写，以及在面试中如何通过这些细节展示你的功底。

我们将代码拆解为五个板块：核心成员变量、拒绝策略、构造方法、核心调度逻辑、工作线程与内部原理。

第一板块：核心成员变量（State & Config）

这一部分定义了线程池的“体质”和“状态”。

// ====================== 七大核心参数 ====================== // 1. 核心线程数：不仅是初始数量，更是“保底数量”。 // 只要线程数没达到这个值，新任务优先创建线程，而不是进队列。 private final int corePoolSize; // 2. 最大线程数：线程池的“物理极限”。 // 当核心满了、队列也满了，才会启用这些“备用名额”。 private final int maximumPoolSize; // 3. & 4. 存活时间与单位：非核心线程的“保质期”。 // 如果非核心线程空闲超过这个时间，就会被回收。 private final long keepAliveTime; private final TimeUnit unit; // 5. 任务阻塞队列：线程池的“缓冲区”。 // 关键点：必须是 BlockingQueue，因为它提供了线程阻塞/唤醒的内置机制（LockSupport）， // 我们不需要自己写复杂的 wait/notify。 private final BlockingQueue<Runnable> workQueue; // 6. 线程工厂：生产线程的地方。 // 作用：可以给线程自定义命名（如 "Order-Thread-1"），方便线上排查故障。 private final ThreadFactory threadFactory; // 7. 拒绝策略：当队列满且线程达上限时的“兜底方案”。 private final RejectHandle rejectHandle; // ====================== 核心状态变量 (并发安全的关键) ====================== // 8. 核心线程计数器 & 9. 非核心线程计数器 // 【面试考点】：为什么要用 AtomicInteger？ // 因为线程的创建（execute中）和销毁（run方法结束时）是异步发生的。 // 虽然 execute 加了锁，但线程结束是在锁外面的。 // 用 int 会导致并发下的 count-- 操作丢失，导致线程数统计不准。 private final AtomicInteger coreThreadCount = new AtomicInteger(0); private final AtomicInteger supportThreadCount = new AtomicInteger(0); // 10. 关闭状态标记 // 【面试考点】：为什么要用 volatile？ // 保证“内存可见性”。当主线程调用 shutdown() 把这个值改为 true 时， // 正在运行的 Worker 线程能通过 CPU 缓存一致性协议，立马看到这个变化， // 从而停止从队列拿任务。如果不加，Worker 可能会一直读旧值（false），导致无法关闭。 private volatile boolean isShutdown = false; // 11. 全局线程池ID生成器 (static) // static 保证所有 MyThreadPool 实例共享这个计数器，用于生成 Pool-1, Pool-2... private static final AtomicInteger poolId = new AtomicInteger(1); // 12. 当前实例ID // final 保证一旦初始化不可变。 private final int currentPoolId;

第二板块：拒绝策略（Strategy Pattern）

这部分展示了策略模式的应用。

// 接口定义：允许用户自定义拒绝逻辑 public interface RejectHandle { void reject(Runnable task, MyThreadPool pool); } // 1. AbortPolicy (JDK默认)：直接抛异常。 // 适用场景：核心业务，不能丢数据，报错让上层知道。 public static final RejectHandle ABORT_POLICY = (task, pool) -> { throw new RejectedExecutionException("..."); }; // 2. CallerRunsPolicy：谁调用的谁执行。 // 逻辑：task.run()。注意这里没有 start()，是直接调用 run，也就是用主线程跑。 // 作用：既不丢任务，又能减缓提交速度（主线程忙着跑任务去了，没空提新任务），形成“背压”。 public static final RejectHandle CALLER_RUNS_POLICY = (task, pool) -> { task.run(); }; // 3. DiscardPolicy：默默丢弃。 // 适用场景：日志收集，丢几条无所谓。 public static final RejectHandle DISCARD_POLICY = (task, pool) -> { // 空实现 }; // 4. DiscardOldestPolicy：喜新厌旧。 // 逻辑：poll() 丢弃队列头部的任务（最早进来的），尝试把新任务塞进去。 public static final RejectHandle DISCARD_OLDEST_POLICY = (task, pool) -> { pool.workQueue.poll(); pool.workQueue.offer(task); };

第三板块：构造方法（Initialization）

public MyThreadPool(...) { // 参数校验（防御性编程）： // 比如核心数不能负，最大数不能小于核心数。 // 面试时提一嘴“参数校验”，说明你有写健壮代码的习惯。 if (corePoolSize < 0 || maximumPoolSize <= 0 || ...) { throw new IllegalArgumentException(...); } // ... 赋值操作 ... // 为当前线程池生成唯一ID this.currentPoolId = poolId.getAndIncrement(); }

第四板块：`execute`方法（核心调度大脑）

这是最重要的部分。请逐行理解。

public void execute(Runnable command) { // 1. 判空：防止空指针炸掉线程池。 if (command == null) throw new NullPointerException(); // 2. 状态检查：如果关门了，直接拒绝。 if (isShutdown) { rejectHandle.reject(command, this); return; } // 【核心锁】：synchronized(this) // 锁定当前线程池对象。 // 目的：保证 "读取线程数 -> 判断 -> 创建线程" 这三步是原子操作。 // 防止多线程并发提交时，瞬间创建出超过 corePoolSize 的线程。 synchronized (this) { int coreCount = coreThreadCount.get(); int supportCount = supportThreadCount.get(); int totalCount = coreCount + supportCount; // --- 步骤 1: 核心线程未满，优先创建 --- if (coreCount < corePoolSize) { createCoreThread(command); // 细节：直接把 task 传给新线程，不进队列。 return; } // --- 步骤 2: 核心已满，尝试入队 --- // 关键点：使用 offer()。 // offer 是非阻塞的！如果队列满返回 false。 // 绝对不能用 put()，否则主线程会卡死在这里等待队列有空位。 if (workQueue.offer(command)) { return; } // --- 步骤 3: 队列已满，尝试创建非核心线程 --- // 只有当前总数 < 最大线程数，才允许创建。 if (totalCount < maximumPoolSize) { createSupportThread(command); // 细节：同样是直接把 task 传给新线程，救火要紧。 return; } // --- 步骤 4: 全满，执行拒绝策略 --- rejectHandle.reject(command, this); } }

第五板块：工作线程与生命周期（Worker）

这里解释了复用和销毁的底层实现。

1. 核心线程逻辑 (`CoreWorker`)

private class CoreWorker implements Runnable { private Runnable firstTask; // 携带的第一个任务 public CoreWorker(Runnable firstTask) { this.firstTask = firstTask; } @Override public void run() { Runnable task = this.firstTask; // 【核心复用逻辑】：死循环 // 条件：task != null (第一次) // || (task = getTaskFromQueue()) != null (后续) // 只要 getTaskFromQueue 不返回 null，这个循环就不结束，线程就不死。 while (task != null || (task = getTaskFromQueue()) != null) { try { task.run(); // 执行真正的业务逻辑 } catch (Throwable e) { // 异常捕获：防止用户代码报错导致核心线程意外终止。 } finally { task = null; // 【重要】：Help GC。执行完必须置空，否则引用链断不开。 } } // 循环退出了？说明线程池关闭了。 coreThreadCount.decrementAndGet(); // 计数器 -1 } } // 获取任务逻辑（阻塞） private Runnable getTaskFromQueue() { try { if (!isShutdown) { // 【关键】：take() 是阻塞方法。 // 队列没数据时，线程 WAITING。 // 它是“核心线程常驻”的根本原因。 return workQueue.take(); } else { // 如果 shutdown 了，就把队列里剩下的取完，取不到就返回 null 让线程死。 return workQueue.poll(); } } catch (InterruptedException e) { // 响应中断（配合 shutdown 方法） return null; } }

2. 非核心线程逻辑 (`SupportWorker`)

private class SupportWorker implements Runnable { // ... 构造同上 ... @Override public void run() { Runnable task = this.firstTask; // 逻辑结构同 CoreWorker，唯一区别是获取任务的方法不一样 while (task != null || (task = getTaskWithTimeout()) != null) { try { task.run(); } catch (Throwable e) { ... } finally { task = null; } } // 退出循环说明超时了或者关闭了，销毁自身。 supportThreadCount.decrementAndGet(); } } // 获取任务逻辑（超时） private Runnable getTaskWithTimeout() { try { if (!isShutdown) { // 【关键】：poll(time, unit) 是超时阻塞。 // 如果 keepAliveTime 时间内没拿到任务，返回 null。 // 返回 null -> while 循环结束 -> run 结束 -> 线程销毁。 // 它是“弹性伸缩”的根本原因。 return workQueue.poll(keepAliveTime, unit); } else { return workQueue.poll(); } } catch (InterruptedException e) { return null; } }

第六板块：创建线程的私有方法

private void createCoreThread(Runnable firstTask) { // 使用 Factory 创建线程 Thread coreThread = threadFactory.newThread(new CoreWorker(firstTask)); // 细节：设置有意义的线程名，方便 debug coreThread.setName("MyThreadPool-" + currentPoolId + "-Core-" + coreThreadCount.incrementAndGet()); coreThread.start(); } // createSupportThread 同理，只是传入的是 SupportWorker

第七板块：优雅关闭 (`shutdown`)

public void shutdown() { // 1. 关门：设置标志位，volatile 保证其他线程立马看见 isShutdown = true; // 2. 喊醒睡觉的人：中断线程 // 为什么？因为很多 Worker 可能正卡在 workQueue.take() 睡觉。 // 如果不 interrupt，它们会一直睡，JVM 就没法退出。 // interrupt 会让 take() 抛出 InterruptedException， // 进而 catch 异常 -> 返回 null -> 退出 run 循环 -> 线程结束。 Thread.currentThread().interrupt(); }

总结：面试怎么复述这一大坨？

你不需要背下每一行代码，你只需要抓住这几个关键词：

七大参数：告诉面试官你懂配置。
synchronized+execute四步走：告诉面试官你懂并发流转。
BlockingQueue 的offervstakevspoll：

- offer：用于生产者，不阻塞（满则 false）。
- take：用于核心消费者，死等（常驻）。
- poll：用于非核心消费者，超时等（销毁）。

Worker 的while循环：告诉面试官你懂线程复用的本质。
State 的安全性：AtomicInteger和volatile的使用。

面试思路

为了深入理解线程池原理，我参照 JDK 的ThreadPoolExecutor手写了这个实现。
它的核心本质是一个多线程环境下的生产者-消费者模型。

生产者是调用execute的外部线程；
消费者是我内部维护的 Worker 线程；
缓冲区是一个BlockingQueue。
为了保证高并发下的准确性，我使用了AtomicInteger记录线程数，用volatile标记关闭状态，用synchronized保证任务提交的原子性。”

第一部分：核心调度逻辑 (`execute`方法) ——这是重点，语速要稳

“关于execute方法，这是线程池的流量入口。为了防止多线程并发提交导致线程数超标，我首先使用synchronized(this)对整个判断逻辑加了锁。也就是‘查状态、判数量、建线程’这几步是原子的。

具体流程我严格遵循了标准的四步走策略：

第一步，优先核心线程：
我先获取AtomicInteger记录的当前核心线程数。如果小于corePoolSize，我直接创建一个CoreWorker。

- 细节：我把当前提交的任务直接传给这个新线程（firstTask），让它启动后立马执行，不走队列，这样响应速度最快。

第二步，尝试入队缓冲：
如果核心线程满了，我尝试把任务放入阻塞队列。

- 关键点：这里我特意使用了队列的offer()方法，而不是put()。
- 理由：因为offer是非阻塞的。如果队列满了，它会立马返回false，这样我就知道该去创建非核心线程了。如果用put，主线程就会卡死在这里，这是绝对不允许的。

第三步，急救扩容（非核心线程）：
如果offer返回 false（队列满了），我判断当前总线程数是否小于maximumPoolSize。如果没满，我就创建一个SupportWorker（非核心线程）来处理积压任务。这实现了线程池的弹性伸缩。
第四步，兜底拒绝：
如果连最大线程数都到了，说明线程池彻底饱和。我就调用RejectHandle接口，执行具体的拒绝策略（比如抛异常），保护系统不被压垮。”

第二部分：线程复用与销毁原理 (`Worker`内部类) ——展示你懂底层

“接下来解释一下线程是怎么复用和销毁的。很多初学者以为线程跑完run就结束了，但我的实现里，Worker类的run方法包含一个while死循环。

这个循环的判断条件是：‘手头有任务，或者能从队列里取到任务’。针对核心和非核心线程，我在取任务的方式上做了区分：

核心线程（CoreWorker）—— 常驻原理：
它调用的是阻塞队列的take()方法。

- 原理：take()是阻塞的。如果队列空了，核心线程会进入WAITING状态挂起，释放 CPU，但不会退出循环，也不会被销毁。一旦有新任务入队，它会被唤醒继续干活。这就是‘核心线程常驻’的秘密。

非核心线程（SupportWorker）—— 销毁原理：
它调用的是阻塞队列的poll(time, unit)方法。

- 原理：这个方法带有超时机制。如果超过了keepAliveTime还没拿到任务，它会返回null。
- 结果：一旦拿到null，while循环条件不满足，循环退出，run方法结束，JVM 就会回收这个线程对象。这就是‘超时销毁’的实现。

健壮性细节：
在循环内部执行task.run()时，我特意包裹了try-catch。这是为了防止用户提交的任务代码抛出 RuntimeException 导致我的 Worker 线程意外退出。捕获异常后，线程可以继续去队列拿下一个任务，保证了线程池的稳定性。”

第三部分：并发安全与优雅关闭 ——加分项

“最后，为了保证代码的健壮性，我在并发细节上也做了处理：

原子计数：
线程数的增减（incrementAndGet/decrementAndGet）我全部使用了AtomicInteger。因为 Worker 线程的结束是异步并发的，用 int 会导致计数丢失，进而导致线程池状态错误。
可见性：
我的isShutdown标志位加了volatile修饰。当外部调用shutdown()时，能保证所有正在运行的 Worker 线程立马看到这个状态变化，停止从队列拿任务。
中断响应：
在shutdown()方法里，我不仅把标志位置为 true，还调用了Thread.currentThread().interrupt()。

- 目的：这是为了唤醒那些正在take()上睡觉的空闲线程，让它们捕获InterruptedException，从而跳出阻塞状态安全退出，避免线程泄露。”

总结

综上所述，我的这个线程池通过synchronized 锁控制入队逻辑，配合阻塞队列的 take/poll 机制控制线程生命周期，实现了一个标准的、线程安全的、支持弹性伸缩的线程池。”

手搓的源码

import java.util.List; import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicIntegerArray; /** * 手写自定义线程池 完美版 * 核心特性：核心线程常驻复用、非核心线程超时销毁、线程安全、优雅关闭、标准拒绝策略、完整七大参数 * 线程复用本质：一个线程对象连续执行多个任务，run方法不结束，持续从队列获取任务 */ public class MyThreadPool { // ====================== 线程池七大核心参数 (完整，私有化+线程安全) ====================== private final int corePoolSize; // 核心线程数-常驻 private final int maximumPoolSize; // 最大线程数-上限(核心+非核心) private final long keepAliveTime; // 非核心线程空闲存活时间 private final TimeUnit unit; // 空闲时间单位 private final BlockingQueue<Runnable> workQueue; // 任务阻塞队列 private final ThreadFactory threadFactory; // 线程工厂-自定义线程创建 private final RejectHandle rejectHandle; // 拒绝策略 // ====================== 线程池核心状态变量 (线程安全+可见性保证) ====================== private final AtomicInteger coreThreadCount = new AtomicInteger(0); // 核心线程数原子计数 (精准，无内存泄漏) private final AtomicInteger supportThreadCount = new AtomicInteger(0); // 非核心线程数原子计数 private volatile boolean isShutdown = false; // 线程池关闭状态，volatile保证可见性 private static final AtomicInteger poolId = new AtomicInteger(1); // 线程池id，用于线程命名 private final int currentPoolId; // 当前线程池的唯一id // ====================== 拒绝策略接口定义 (必须实现) ====================== public interface RejectHandle { /** * 任务拒绝的回调方法 * @param task 被拒绝的任务 * @param pool 当前线程池对象 */ void reject(Runnable task, MyThreadPool pool); } // ====================== 内置默认拒绝策略 (和JDK原生对齐，4种可选) ====================== public static final RejectHandle ABORT_POLICY = (task, pool) -> { throw new RejectedExecutionException("任务[" + task + "]被拒绝，线程池已饱和！"); }; public static final RejectHandle CALLER_RUNS_POLICY = (task, pool) -> { // 调用者线程执行任务，回退策略 task.run(); }; public static final RejectHandle DISCARD_POLICY = (task, pool) -> { // 静默丢弃，无任何操作 }; public static final RejectHandle DISCARD_OLDEST_POLICY = (task, pool) -> { // 丢弃队列最旧的任务，再尝试入队当前任务 pool.workQueue.poll(); pool.workQueue.offer(task); }; // ====================== 线程池构造方法 (重载2个，满足不同使用场景，必传核心参数) ====================== public MyThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) { this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), ABORT_POLICY); } public MyThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectHandle rejectHandle) { // 参数合法性校验，杜绝非法参数 if (corePoolSize < 0 || maximumPoolSize <= 0 || maximumPoolSize < corePoolSize || keepAliveTime < 0) { throw new IllegalArgumentException("线程池参数不合法！"); } if (workQueue == null || threadFactory == null || rejectHandle == null) { throw new NullPointerException("线程池核心组件不能为空！"); } this.corePoolSize = corePoolSize; this.maximumPoolSize = maximumPoolSize; this.keepAliveTime = keepAliveTime; this.unit = unit; this.workQueue = workQueue; this.threadFactory = threadFactory; this.rejectHandle = rejectHandle; this.currentPoolId = poolId.getAndIncrement(); } // ====================== 核心方法：提交任务 (标准线程池流程，加锁保证原子性，线程安全) ====================== public void execute(Runnable command) { // 1. 任务为空直接抛异常 if (command == null) { throw new NullPointerException("任务不能为空！"); } // 2. 线程池已关闭，直接拒绝 if (isShutdown) { rejectHandle.reject(command, this); return; } // 加锁保证【线程数判断+创建线程】的原子性，锁对象是当前线程池，全局唯一锁，解决并发安全问题 synchronized (this) { int coreCount = coreThreadCount.get(); int supportCount = supportThreadCount.get(); int totalCount = coreCount + supportCount; // 步骤1：核心线程数未满，创建核心线程执行任务 (优先级最高) if (coreCount < corePoolSize) { createCoreThread(command); return; } // 步骤2：核心线程已满，尝试将任务加入阻塞队列，成功则排队 if (workQueue.offer(command)) { return; } // 步骤3：队列已满，判断是否达到最大线程数，未达到则创建非核心线程 if (totalCount < maximumPoolSize) { createSupportThread(command); return; } // 步骤4：当前核心线程满+队列队列满+线程数满，执行拒绝策略 (兜底) rejectHandle.reject(command, this); } } // ====================== 私有方法：创建核心线程 (常驻线程，无限复用，无超时) ====================== private void createCoreThread(Runnable firstTask) { Thread coreThread = threadFactory.newThread(new CoreWorker(firstTask)); coreThread.setName("MyThreadPool-" + currentPoolId + "-CoreThread-" + coreThreadCount.incrementAndGet()); coreThread.start(); } // ====================== 私有方法：创建非核心线程 (临时工，超时销毁，弹性伸缩) ====================== private void createSupportThread(Runnable firstTask) { Thread supportThread = threadFactory.newThread(new SupportWorker(firstTask)); supportThread.setName("MyThreadPool-" + currentPoolId + "-SupportThread-" + supportThreadCount.incrementAndGet()); supportThread.start(); } // ====================== 核心工作线程：实现线程复用的核心逻辑 (内部类，私有化) ====================== private class CoreWorker implements Runnable { private Runnable firstTask; public CoreWorker(Runnable firstTask) { this.firstTask = firstTask; } @Override public void run() { Runnable task = this.firstTask; // 核心逻辑：死循环持续获取任务，执行完一个再拿一个，实现线程复用 // 刚创建时处理手头的新任务，之后去队列里抢任务 while (task != null || (task = getTaskFromQueue()) != null) { try { task.run(); // 执行任务 } catch (Throwable e) { // 捕获所有异常，避免核心线程因为任务异常而死亡，线程池失效 System.err.println("核心线程执行任务异常：" + e.getMessage()); } finally { task = null; // 释放任务引用，便于GC } } // 线程池关闭后，核心线程退出，原子计数减1 coreThreadCount.decrementAndGet(); } } // ====================== 非核心工作线程：超时销毁的核心逻辑 (内部类，私有化) ====================== private class SupportWorker implements Runnable { private Runnable firstTask; public SupportWorker(Runnable firstTask) { this.firstTask = firstTask; } @Override public void run() { Runnable task = this.firstTask; // 核心逻辑：超时拿不到任务就退出，实现空闲销毁；拿到任务就执行，实现线程复用 // 刚创建时处理手头的新任务，之后去队列里抢任务。 while (task != null || (task = getTaskWithTimeout()) != null) { try { task.run(); // 执行任务 } catch (Throwable e) { System.err.println("非核心线程执行任务异常：" + e.getMessage()); } finally { task = null; // 释放任务引用 } } // 超时退出/线程池关闭，非核心线程计数减1，解决内存泄漏+计数不准问题 supportThreadCount.decrementAndGet(); } } // ====================== 私有方法：核心线程从队列拿任务 (阻塞式，拿不到一直等) ====================== private Runnable getTaskFromQueue() { try { // 线程池未关闭，阻塞获取任务 if (!isShutdown) { // 去阻塞队列里面阻塞获取任务,注意这个是阻塞的 return workQueue.take(); } else { // 线程池已关闭，尝试拿剩余任务，拿不到就退出 return workQueue.poll(); } } catch (InterruptedException e) { // 响应中断，线程池关闭时触发，优雅退出 Thread.currentThread().interrupt(); return null; } } // ====================== 私有方法：非核心线程从队列拿任务 (超时式，拿不到就退出) ====================== private Runnable getTaskWithTimeout() { try { if (!isShutdown) { // 超时获取任务，超时返回null，线程就会退出 return workQueue.poll(keepAliveTime, unit); } else { // 线程池已关闭，尝试拿剩余任务，拿不到就退出 return workQueue.poll(); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); return null; } } // ====================== 优雅关闭线程池 (核心方法，必须实现) ====================== public void shutdown() { isShutdown = true; // 中断所有线程，让阻塞的线程响应中断，快速退出 Thread.currentThread().interrupt(); System.out.println("线程池已关闭，不再接收新任务，正在执行剩余任务..."); } // ====================== 辅助方法：获取线程池状态 ====================== public int getCorePoolSize() { return corePoolSize; } public int getMaximumPoolSize() { return maximumPoolSize; } public int getCoreThreadCount() { return coreThreadCount.get(); } public int getSupportThreadCount() { return supportThreadCount.get(); } public int getQueueSize() { return workQueue.size(); } }

拒绝策略已简写

public interface RejectHandle { void reject(Runnable task,MyThreadPool myThreadPool); }