- 正确答案：线程池是一种并发编程中用于管理和复用线程的资源池，其核心由任务队列（阻塞队列）、工作线程集合、线程池状态控制（运行/关闭/终止）、拒绝策略和线程工厂等组成。一个典型的Java风格线程池（如ThreadPoolExecutor）可手动实现为：维护一个BlockingQueue作为待执行任务队列；固定数量的工作线程持续从队列中take()任务并执行；支持动态调整核心线程数、最大线程数、空闲线程存活时间；具备shutdown()（不再接受新任务，但执行完已有任务）和shutdownNow()（尝试中断所有正在执行的任务，并返回未执行任务列表）语义；当任务提交时若队列已满且线程数已达maximumPoolSize，则触发拒绝策略（如AbortPolicy抛异常、CallerRunsPolicy由提交线程自己执行等）。 - 解答思路： 1. 明确线程池的关键组件：任务容器（需线程安全+阻塞能力）、执行者（Worker线程）、生命周期管理（状态机）、拒绝机制（背压控制）。 2. 设计状态变量（如RUNNING、SHUTDOWN、STOP、TERMINATED），保证状态变更的原子性和可见性（使用volatile + CAS或AtomicInteger）。 3. Worker线程需具备自我管理能力：循环获取任务→执行→异常处理→空闲超时回收（需结合Thread.interrupt()与park/unpark或wait/notify）。 4. 任务提交逻辑需考虑线程安全（offer/put操作本身线程安全，但整体流程需避免竞态，如“先检查线程数再创建线程”需加锁或CAS重试）。 5. shutdown逻辑需双重保障：禁止新任务入队 + 中断空闲线程 + 等待活跃线程自然结束；shutdownNow需遍历所有Worker并interrupt()，同时清空队列。 - 深度知识讲解：底层依赖的核心数据结构与机制包括：（1）阻塞队列：推荐使用LinkedBlockingQueue（基于链表+两把锁：putLock/takeLock，读写分离）或ArrayBlockingQueue（基于数组+单ReentrantLock+两个Condition，内存连续但容量固定）。其阻塞语义本质是通过AQS的Condition.await()实现线程挂起，由生产者调用signal()唤醒消费者。（2）线程复用原理：每个Worker封装一个Thread对象和一个while循环体，该循环在RUNNING状态下不断调用queue.poll(keepAliveTime, TimeUnit)或take()，避免频繁创建销毁线程带来的JVM开销（包括栈内存分配、GC压力、上下文切换成本）。（3）线程中断机制：interrupt()并非强制终止线程，而是设置线程的中断标志位，并在以下场景触发响应：① 阻塞在wait()/join()/sleep()时抛InterruptedException；② 阻塞在InterruptibleChannel（如NIO SocketChannel）时关闭通道并抛ClosedByInterruptException；③ 调用Thread.interrupted()检测并清除标志位。因此Worker必须在catch InterruptedException后主动退出循环，或在执行任务前检查Thread.currentThread().isInterrupted()。（4）AQS（AbstractQueuedSynchronizer）：ThreadPoolExecutor内部虽未直接继承AQS，但其状态控制（ctl变量：高3位表示状态，低29位表示workerCount）采用CAS原子更新（Unsafe.compareAndSwapInt），属于AQS思想的轻量级应用；而阻塞队列底层大量使用AQS的独占/共享模式实现同步。（5）拒绝策略本质是系统过载保护（Back Pressure），对应流控模型中的“lossy”策略。更高级方案可结合令牌桶（Guava RateLimiter）或滑动窗口限流，在提交端前置拦截，而非仅靠队列满后抛异常。 - 伪代码（简化版，含核心逻辑）： ``` class SimpleThreadPool { private final BlockingQueue workQueue; private final List workers; private final ThreadFactory threadFactory; private final RejectedExecutionHandler handler; private volatile int ctl = RUNNING << COUNT_BITS; // 32位整数，高3位=state，低29位=workerCount private static final int COUNT_BITS = 29; private static final int CAPACITY = (1 << COUNT_BITS) - 1; private static final int RUNNING = -1 << COUNT_BITS; private static final int SHUTDOWN = 0 << COUNT_BITS; private static final int STOP = 1 << COUNT_BITS; private static final int TIDYING = 2 << COUNT_BITS; private static final int TERMINATED = 3 << COUNT_BITS; public SimpleThreadPool(int corePoolSize, int maxPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) { this.workQueue = workQueue; this.workers = new ArrayList<>(); this.threadFactory = threadFactory; this.handler = handler; // 初始化corePoolSize个核心线程 for (int i = 0; i < corePoolSize; i++) { addWorker(null, true); } } private boolean addWorker(Runnable firstTask, boolean core) { retry: for (;;) { int c = ctl.get(); int rs = runStateOf(c); if (rs >= SHUTDOWN && (rs != SHUTDOWN || firstTask != null || workQueue.isEmpty())) return false; for (;;) { int wc = workerCountOf(c); if (wc >= (core ? corePoolSize : maximumPoolSize)) return false; if (ctl.compareAndSet(c, c + 1)) { break retry; } c = ctl.get(); } } Worker w = new Worker(firstTask); Thread t = threadFactory.newThread(w); if (t != null) { workers.add(w); t.start(); return true; } return false; } public void execute(Runnable command) { if (command == null) throw new NullPointerException(); int c = ctl.get(); if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) return; c = ctl.get(); } if (isRunning(c) && workQueue.offer(command)) { if (!isRunning(ctl.get()) && remove(command)) reject(command); } else if (!addWorker(command, false)) reject(command); } private void reject(Runnable command) { handler.rejectedExecution(command, this); } private final class Worker implements Runnable { final Thread thread; Runnable firstTask; volatile long completedTasks; Worker(Runnable firstTask) { this.firstTask = firstTask; this.thread = getThreadFactory().newThread(this); } public void run() { runWorker(this); } } final void runWorker(Worker w) { Thread wt = Thread.currentThread(); Runnable task = w.firstTask; w.firstTask = null; boolean completedAbruptly = true; try { while (task != null || (task = getTask()) != null) { try { task.run(); } finally { task = null; w.completedTasks++; } } completedAbruptly = false; } finally { processWorkerExit(w, completedAbruptly); } } private Runnable getTask() { boolean timedOut = false; for (;;) { int c = ctl.get(); int rs = runStateOf(c); if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { decrementWorkerCount(); return null; } int wc = workerCountOf(c); boolean timed = wc > corePoolSize || allowCoreThreadTimeOut; if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) { if (compareAndDecrementWorkerCount(c)) return null; continue; } try { Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take(); if (r != null) return r; timedOut = true; } catch (InterruptedException retry) { timedOut = false; } } } private void processWorkerExit(Worker w, boolean completedAbruptly) { if (completedAbruptly) decrementWorkerCount(); final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { completedTaskCount += w.completedTasks; workers.remove(w); } finally { mainLock.unlock(); } tryTerminate(); } public void shutdown() { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { checkShutdownAccess(); advanceRunState(SHUTDOWN); interruptIdleWorkers(); onShutdown(); } finally { mainLock.unlock(); } tryTerminate(); } public List shutdownNow() { List tasks; final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { checkShutdownAccess(); advanceRunState(STOP); interruptWorkers(); tasks = drainQueue(); } finally { mainLock.unlock(); } tryTerminate(); return tasks; } private void interruptIdleWorkers() { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { for (Worker w : workers) w.interruptIfIdle(); } finally { mainLock.unlock(); } } private void interruptWorkers() { for (Worker w : workers) w.interruptNow(); } private List drainQueue() { BlockingQueue q = workQueue; List taskList = new ArrayList<>(); q.drainTo(taskList); if (!q.isEmpty()) { for (Runnable r : q.toArray(new Runnable[0])) { if (q.remove(r)) taskList.add(r); } } return taskList; } private void tryTerminate() { for (;;) { int c = ctl.get(); if (isRunning(c) || runStateAtLeast(c, TIDYING) || (runStateOf(c) == SHUTDOWN && !workQueue.isEmpty())) { return; } if (workerCountOf(c) != 0) { interruptIdleWorkers(); return; } final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { if (ctl.compareAndSet(c, TIDYING << COUNT_BITS)) { try { terminated(); } finally { ctl.set(TERMINATED << COUNT_BITS); termination.signalAll(); } return; } } finally { mainLock.unlock(); } } } } ``` 补充说明：以上伪代码高度浓缩了JDK ThreadPoolExecutor的核心逻辑，省略了部分细节（如mainLock、terminated()钩子方法、统计字段的并发更新等），但保留了状态机转换、Worker生命周期、任务获取阻塞/超时、拒绝策略触发点、shutdown/shutdownNow语义差异等关键路径。实际工业级实现还需考虑： - 线程局部存储（ThreadLocal）清理（防止内存泄漏）； - ForkJoinPool的work-stealing机制（适用于递归分治任务）； - Netty EventLoopGroup的单线程绑定模型（I/O线程不切换，避免锁竞争）； - Spring TaskExecutor抽象对ThreadPoolTaskExecutor的封装，支持@Async注解驱动。

精选高频手撕题目合集

全站热榜

创作者周榜