Java并发——线程池

前言在Java并发编程的世界里线程池是一个绕不开的核心话题。无论是面试中的高频考点还是实际项目中的性能优化线程池都扮演着至关重要的角色。本文将带你深入理解线程池的设计思想、实现原理以及最佳实践。一、为什么需要线程池在了解线程池之前我们先思考一个问题为什么不能每次需要执行任务时就创建一个新线程1.1 传统方式的弊端// 传统方式 - 每次任务都创建新线程 new Thread(() - { // 执行任务 doSomething(); }).start();这种方式存在以下问题资源消耗高线程的创建和销毁需要消耗系统资源稳定性差无限制创建线程可能导致系统资源耗尽OOM管理困难无法统一管理线程的生命周期1.2 线程池的优势线程池通过复用线程带来了以下好处降低资源消耗复用已创建的线程减少创建和销毁的开销提高响应速度任务到达时无需等待线程创建即可执行提高线程可管理性统一分配、调优和监控二、线程池的核心设计2.1 线程池的关键参数Java中的线程池通过ThreadPoolExecutor类实现其构造函数包含7个核心参数public ThreadPoolExecutor( int corePoolSize, // 核心线程数 int maximumPoolSize, // 最大线程数 long keepAliveTime, // 空闲线程存活时间 TimeUnit unit, // 时间单位 BlockingQueueRunnable workQueue, // 任务队列 ThreadFactory threadFactory, // 线程工厂 RejectedExecutionHandler handler // 拒绝策略 )2.2 参数详解corePoolSize核心线程数核心线程会一直存活即使空闲当线程数小于核心线程数时即使有空闲线程也会创建新线程maximumPoolSize最大线程数线程池允许创建的最大线程数当任务队列满了且线程数小于最大线程数时会创建新线程workQueue任务队列常用的任务队列有SynchronousQueue不存储元素的阻塞队列每个插入必须等待一个移除LinkedBlockingQueue无界队列默认大小Integer.MAX_VALUEArrayBlockingQueue有界队列需要指定容量keepAliveTime空闲存活时间非核心线程空闲时的存活时间如果设置了allowCoreThreadTimeOut(true)核心线程也会被回收handler拒绝策略当线程池无法处理新任务时的处理策略AbortPolicy默认抛出RejectedExecutionExceptionCallerRunsPolicy在调用者线程中运行任务DiscardPolicy静默丢弃任务DiscardOldestPolicy丢弃队列头部的任务重试提交2.3 线程池的状态线程池有5种运行状态// 线程池状态 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; // 终止状态转换图RUNNING - SHUTDOWN - TIDYING - TERMINATED RUNNING - STOP - TIDYING - TERMINATED三、线程池的工作流程3.1 任务提交流程public void execute(Runnable command) { int c ctl.get(); // 1. 当前线程数 核心线程数创建新线程 if (workerCountOf(c) corePoolSize) { if (addWorker(command, true)) return; c ctl.get(); } // 2. 核心线程已满尝试加入队列 if (isRunning(c) workQueue.offer(command)) { // 双重检查 int recheck ctl.get(); if (!isRunning(recheck) remove(command)) reject(command); else if (workerCountOf(recheck) 0) addWorker(null, false); } // 3. 队列已满创建非核心线程 else if (!addWorker(command, false)) reject(command); // 4. 超过最大线程数执行拒绝策略 }3.2 图解工作流程提交任务 ↓ 线程数 核心线程数 ├─ 是 → 创建核心线程执行任务 └─ 否 → 任务队列是否已满 ├─ 否 → 加入任务队列等待 └─ 是 → 线程数 最大线程数 ├─ 是 → 创建非核心线程执行任务 └─ 否 → 执行拒绝策略四、常用线程池的实现4.1 Executors工厂类提供的线程池1. newFixedThreadPoolpublic static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor( nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueueRunnable() ); }固定大小的线程池使用无界队列可能导致内存溢出适合执行长期任务2. newCachedThreadPoolpublic static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor( 0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueueRunnable() ); }核心线程数为0最大线程数无限适合执行大量短期任务线程空闲60秒后回收3. newSingleThreadExecutorpublic static ExecutorService newSingleThreadExecutor() { return new FinalizableDelegatedExecutorService( new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueueRunnable()) ); }单线程的线程池保证任务按顺序执行使用无界队列4. newScheduledThreadPoolpublic static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) { return new ScheduledThreadPoolExecutor(corePoolSize); }支持定时和周期性任务基于DelayQueue实现4.2 为什么不推荐使用Executors《阿里巴巴Java开发手册》明确指出线程池不允许使用Executors创建而是通过ThreadPoolExecutor的方式。原因FixedThreadPool和SingleThreadPool使用无界队列任务堆积可能导致OOMCachedThreadPool允许创建大量线程可能导致OOM五、最佳实践5.1 如何合理配置线程池线程池的配置需要根据任务类型来确定CPU密集型任务// 线程数 CPU核心数 1 int corePoolSize Runtime.getRuntime().availableProcessors() 1;IO密集型任务// 线程数 CPU核心数 * (1 平均等待时间/平均工作时间) int corePoolSize Runtime.getRuntime().availableProcessors() * 2;混合型任务可以将任务拆分为CPU密集型和IO密集型使用不同的线程池分别处理5.2 自定义线程池示例Configuration public class ThreadPoolConfig { Bean(customThreadPool) public ExecutorService customThreadPool() { // 获取CPU核心数 int cpuCores Runtime.getRuntime().availableProcessors(); ThreadPoolExecutor executor new ThreadPoolExecutor( cpuCores, // 核心线程数 cpuCores * 2, // 最大线程数 60, // 空闲存活时间 TimeUnit.SECONDS, new ArrayBlockingQueue(200), // 有界队列 new CustomThreadFactory(), // 自定义线程工厂 new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略 ); // 允许回收核心线程 executor.allowCoreThreadTimeOut(true); return executor; } static class CustomThreadFactory implements ThreadFactory { private final AtomicInteger threadNumber new AtomicInteger(1); Override public Thread newThread(Runnable r) { Thread t new Thread(r, custom-thread- threadNumber.getAndIncrement()); t.setDaemon(false); t.setPriority(Thread.NORM_PRIORITY); return t; } } }5.3 监控线程池Component public class ThreadPoolMonitor { Autowired private ThreadPoolExecutor threadPool; Scheduled(fixedDelay 5000) public void monitor() { log.info(线程池状态监控:); log.info(核心线程数: {}, threadPool.getCorePoolSize()); log.info(活跃线程数: {}, threadPool.getActiveCount()); log.info(最大线程数: {}, threadPool.getMaximumPoolSize()); log.info(当前线程数: {}, threadPool.getPoolSize()); log.info(任务队列大小: {}, threadPool.getQueue().size()); log.info(已完成任务数: {}, threadPool.getCompletedTaskCount()); log.info(总任务数: {}, threadPool.getTaskCount()); } }六、进阶话题6.1 线程池的关闭优雅关闭线程池的流程public void shutdownThreadPool(ExecutorService threadPool) { // 不再接收新任务 threadPool.shutdown(); try { // 等待60秒让已有任务完成 if (!threadPool.awaitTermination(60, TimeUnit.SECONDS)) { // 超时后强制关闭 threadPool.shutdownNow(); // 再等待60秒 if (!threadPool.awaitTermination(60, TimeUnit.SECONDS)) { log.error(线程池未能正常关闭); } } } catch (InterruptedException e) { threadPool.shutdownNow(); Thread.currentThread().interrupt(); } }6.2 动态调整线程池Component public class DynamicThreadPool { private final ThreadPoolExecutor executor; public DynamicThreadPool() { this.executor new ThreadPoolExecutor( 10, 20, 60, TimeUnit.SECONDS, new LinkedBlockingQueue(1000) ); } // 动态调整核心线程数 public void setCorePoolSize(int size) { executor.setCorePoolSize(size); } // 动态调整最大线程数 public void setMaximumPoolSize(int size) { executor.setMaximumPoolSize(size); } // 动态调整拒绝策略 public void setRejectedHandler(RejectedExecutionHandler handler) { executor.setRejectedExecutionHandler(handler); } }七、常见问题与解决方案7.1 线程池任务执行异常处理public class ExceptionHandlingThreadPool { public static void main(String[] args) { ThreadPoolExecutor executor new ThreadPoolExecutor( 1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue() ); // 方式1使用Future获取异常 Future? future executor.submit(() - { throw new RuntimeException(任务异常); }); try { future.get(); // 这里会抛出异常 } catch (Exception e) { log.error(任务执行异常, e); } // 方式2自定义ThreadFactory设置UncaughtExceptionHandler ThreadFactory factory r - { Thread t new Thread(r); t.setUncaughtExceptionHandler((thread, throwable) - { log.error(线程 {} 发生异常, thread.getName(), throwable); }); return t; }; // 方式3重写afterExecute方法 ThreadPoolExecutor executor2 new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue()) { Override protected void afterExecute(Runnable r, Throwable t) { super.afterExecute(r, t); if (t null r instanceof Future?) { try { ((Future?) r).get(); } catch (Exception e) { t e; } } if (t ! null) { log.error(任务执行异常, t); } } }; } }7.2 线程池死锁问题// 错误示例可能造成死锁 ExecutorService executor Executors.newSingleThreadExecutor(); executor.submit(() - { FutureString future executor.submit(() - task); return future.get(); // 等待子任务完成但子任务无法执行 }); // 正确做法使用不同线程池 ExecutorService parentExecutor Executors.newFixedThreadPool(2); ExecutorService childExecutor Executors.newCachedThreadPool();八、总结线程池是Java并发编程中的重要工具正确使用线程池能够显著提升系统性能和稳定性。本文从线程池的设计思想出发深入讲解了其工作原理、参数配置、最佳实践以及常见问题。核心要点回顾理解线程池的7个核心参数及其作用掌握线程池的工作流程和状态转换根据任务类型合理配置线程池参数使用ThreadPoolExecutor创建线程池避免使用Executors实现完善的监控和异常处理机制