当前位置： 首页 > java >正文

Java线程池

java 2025/6/23 19:05:28

文章目录

      • 线程数量配置
      • 队列选择策略
      • 监控和调优
    • 线程池的核心优势
      • 性能优化
      • 资源控制
      • 响应速度提升
      • 统一管理
    • Java线程池的实现架构
      • ThreadPoolExecutor类
      • 任务队列机制
      • 拒绝策略
    • 线程池的工作流程
    • 常见线程池类型
      • FixedThreadPool
      • CachedThreadPool
      • SingleThreadExecutor
      • ScheduledThreadPool
    • 最佳实践与配置建议
      • 线程数量配置
      • 队列选择策略
      • 监控和调优
      • 异常处理
      • 优雅关闭
    • 性能优化技巧
      • 预热策略
      • 任务分解
      • 线程本地变量
    • 常见问题与解决方案
      • 内存泄漏
      • 死锁问题
      • 性能瓶颈

线程数量配置

线程池大小的配置需要根据具体的应用场景来确定。对于CPU密集型任务，线程数量通常设置为CPU核心数加1，这样能够充分利用CPU资源而不会产生过多的上下文切换开销。

对于IO密集型任务，由于线程经常处于等待状态，可以设置更多的线程数量。一般建议设置为CPU核心数的2倍或更多，具体数值需要通过性能测试来确定最优值。

public class ThreadPoolSizingExample {public static void main(String[] args) throws InterruptedException {int cpuCores = Runtime.getRuntime().availableProcessors();System.out.println("CPU核心数: " + cpuCores);// CPU密集型任务线程池配置demonstrateCpuIntensivePool(cpuCores);Thread.sleep(2000);// IO密集型任务线程池配置demonstrateIoIntensivePool(cpuCores);}private static void demonstrateCpuIntensivePool(int cpuCores) throws InterruptedException {System.out.println("\n=== CPU密集型任务线程池 ===");// CPU密集型：核心数 + 1ThreadPoolExecutor cpuPool = new ThreadPoolExecutor(cpuCores + 1,cpuCores + 1,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<>(50),new ThreadFactory() {private AtomicInteger threadNum = new AtomicInteger(1);@Overridepublic Thread newThread(Runnable r) {return new Thread(r, "CPU-Worker-" + threadNum.getAndIncrement());}});long startTime = System.currentTimeMillis();CountDownLatch latch = new CountDownLatch(10);for (int i = 0; i < 10; i++) {final int taskId = i;cpuPool.submit(() -> {System.out.println("CPU密集任务" + taskId + "开始 - " + Thread.currentThread().getName());// 模拟CPU密集型计算long result = 0;for (long j = 0; j < 100_000_000L; j++) {result += j;}System.out.println("CPU密集任务" + taskId + "完成 - 结果: " + result);latch.countDown();});}latch.await();long cpuTime = System.currentTimeMillis() - startTime;System.out.println("CPU密集型任务总耗时: " + cpuTime + "ms");cpuPool.shutdown();}private static void demonstrateIoIntensivePool(int cpuCores) throws InterruptedException {System.out.println("\n=== IO密集型任务线程池 ===");// IO密集型：核心数 * 2ThreadPoolExecutor ioPool = new ThreadPoolExecutor(cpuCores * 2,cpuCores * 2,60L, TimeUnit.SECONDS,new LinkedBlockingQueue<>(100),new ThreadFactory() {private AtomicInteger threadNum = new AtomicInteger(1);@Overridepublic Thread newThread(Runnable r) {return new Thread(r, "IO-Worker-" + threadNum.getAndIncrement());}});long startTime = System.currentTimeMillis();CountDownLatch latch = new CountDownLatch(20);for (int i = 0; i < 20; i++) {final int taskId = i;ioPool.submit(() -> {System.out.println("IO密集任务" + taskId + "开始 - " + Thread.currentThread().getName());try {// 模拟IO操作（网络请求、文件读写等）Thread.sleep(500);// 模拟少量CPU计算int result = taskId * 2;System.out.println("IO密集任务" + taskId + "完成 - 结果: " + result);} catch (InterruptedException e) {Thread.currentThread().interrupt();}latch.countDown();});}latch.await();long ioTime = System.currentTimeMillis() - startTime;System.out.println("IO密集型任务总耗时: " + ioTime + "ms");ioPool.shutdown();}
}

队列选择策略

工作队列的选择对线程池性能有重要影响。有界队列能够防止内存溢出，但可能导致任务被拒绝。无界队列避免了任务拒绝问题，但可能在高负载时消耗大量内存。

在实际应用中，建议使用有界队列并设置合理的容量。队列大小可以根据系统的内存容量和任务处理速度来确定。

public class QueueSelectionExample {public static void main(String[] args) throws InterruptedException {System.out.println("=== 队列选择策略演示 ===");// 1. 有界队列 - 适合内存敏感应用demonstrateBoundedQueue();Thread.sleep(2000);// 2. 无界队列 - 适合任务不能丢失的场景demonstrateUnboundedQueue();Thread.sleep(2000);// 3. 队列容量计算示例demonstrateQueueSizeCalculation();}private static void demonstrateBoundedQueue() {System.out.println("\n--- 有界队列演示 ---");ThreadPoolExecutor boundedPool = new ThreadPoolExecutor(2, 4, 60L, TimeUnit.SECONDS,new ArrayBlockingQueue<>(5), // 有界队列，容量为5new ThreadPoolExecutor.CallerRunsPolicy());for (int i = 1; i <= 12; i++) {final int taskId = i;try {boundedPool.submit(() -> {System.out.println("有界队列任务" + taskId + " - " + Thread.currentThread().getName());try {Thread.sleep(1000);} catch (InterruptedException e) {Thread.currentThread().interrupt();}});System.out.println("任务" + taskId + "提交成功");} catch (Exception e) {System.out.println("任务" + taskId + "提交失败: " + e.getMessage());}}boundedPool.shutdown();}private static void demonstrateUnboundedQueue() {System.out.println("\n--- 无界队列演示 ---");ThreadPoolExecutor unboundedPool = new ThreadPoolExecutor(2, 2, 0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<>() // 无界队列);// 提交大量任务for (int i = 1; i <= 50; i++) {final int taskId = i;unboundedPool.submit(() -> {try {Thread.sleep(100);if (taskId % 10 == 0) {System.out.println("无界队列任务" + taskId + "执行完成");}} catch (InterruptedException e) {Thread.currentThread().interrupt();}});}System.out.println("无界队列任务提交完成，队列大小: " + unboundedPool.getQueue().size());unboundedPool.shutdown();}private static void demonstrateQueueSizeCalculation() {System.out.println("\n--- 队列容量计算演示 ---");// 计算合理的队列大小int corePoolSize = 4;int avgTaskDurationMs = 1000; // 平均任务执行时间1秒int tasksPerSecond = 10;      // 每秒任务提交量// 队列容量 = 每秒任务提交量 * 平均任务时间 - 核心线程数int recommendedQueueSize = Math.max(1, tasksPerSecond * (avgTaskDurationMs / 1000) - corePoolSize);System.out.println("推荐队列容量计算:");System.out.println("核心线程数: " + corePoolSize);System.out.println("平均任务执行时间: " + avgTaskDurationMs + "ms");System.out.println("每秒任务提交量: " + tasksPerSecond);System.out.println("推荐队列容量: " + recommendedQueueSize);ThreadPoolExecutor calculatedPool = new ThreadPoolExecutor(corePoolSize,corePoolSize * 2,60L, TimeUnit.SECONDS,new ArrayBlockingQueue<>(recommendedQueueSize),new ThreadPoolExecutor.CallerRunsPolicy());// 测试计算出的队列容量for (int i = 1; i <= 20; i++) {final int taskId = i;calculatedPool.submit(() -> {try {Thread.sleep(avgTaskDurationMs);System.out.println("计算队列任务" + taskId + "完成");} catch (InterruptedException e) {Thread.currentThread().interrupt();}});}calculatedPool.shutdown();}
}

监控和调优

线程池的运行状态需要持续监控以确保最优性能。重要的监控指标包括活跃线程数、队列长度、任务完成数量和拒绝任务数量。通过这些指标可以判断线程池配置是否合理。

当发现线程池性能不佳时，可以通过调整核心线程数、最大线程数、队列大小等参数来优化性能。调优过程需要结合实际的业务负载情况进行测试验证。

public class ThreadPoolMonitoringExample {private static final AtomicLong rejectedTaskCount = new AtomicLong(0);public static void main(String[] args) throws InterruptedException {// 创建带监控的线程池ThreadPoolExecutor monitoredPool = createMonitoredThreadPool();// 启动监控startMonitoring(monitoredPool);// 模拟不同负载情况simulateVaryingLoad(monitoredPool);// 等待任务完成Thread.sleep(30000);monitoredPool.shutdown();}private static ThreadPoolExecutor createMonitoredThreadPool() {return new ThreadPoolExecutor(3,    // 核心线程数6,    // 最大线程数30L,  // 空闲线程存活时间TimeUnit.SECONDS,new ArrayBlockingQueue<>(10), // 有界队列new MonitoredThreadFactory(),new MonitoredRejectionHandler());}private static void startMonitoring(ThreadPoolExecutor executor) {ScheduledExecutorService monitor = Executors.newSingleThreadScheduledExecutor();monitor.scheduleAtFixedRate(() -> {System.out.printf("%s - 监控报告:%n", new Date());System.out.printf("  核心线程数: %d%n", executor.getCorePoolSize());System.out.printf("  当前线程数: %d%n", executor.getPoolSize());System.out.printf("  活跃线程数: %d%n", executor.getActiveCount());System.out.printf("  历史最大线程数: %d%n",# Java线程池技术详解：高效并发编程的核心工具## 引言在现代Java应用程序开发中，线程池已成为处理并发任务的标准解决方案。线程池不仅能够有效管理系统资源，还能显著提升应用程序的性能和响应能力。本文将深入探讨Java线程池的核心原理、实现机制以及最佳实践，为开发人员提供全面的技术指导。## 什么是线程池线程池是一种基于池化技术的线程管理机制，它预先创建并维护一定数量的工作线程，用于执行提交的任务。当任务到达时，线程池会从现有线程中分配一个空闲线程来处理该任务，而不是为每个任务创建新的线程。这种设计模式有效解决了频繁创建和销毁线程所带来的性能开销问题。```java
// 传统的线程创建方式 - 每个任务创建新线程
public class TraditionalThreadExample {public static void main(String[] args) {for (int i = 0; i < 10; i++) {final int taskId = i;// 每次都创建新线程，开销大new Thread(() -> {System.out.println("执行任务 " + taskId + " - 线程: " + Thread.currentThread().getName());try {Thread.sleep(1000);} catch (InterruptedException e) {Thread.currentThread().interrupt();}}).start();}}
}// 使用线程池的方式 - 复用线程
public class ThreadPoolExample {public static void main(String[] args) {ExecutorService executor = Executors.newFixedThreadPool(3);for (int i = 0; i < 10; i++) {final int taskId = i;// 复用现有线程，减少创建销毁开销executor.submit(() -> {System.out.println("执行任务 " + taskId + " - 线程: " + Thread.currentThread().getName());try {Thread.sleep(1000);} catch (InterruptedException e) {Thread.currentThread().interrupt();}});}executor.shutdown();}
}

线程池的核心价值在于资源复用和统一管理。通过复用现有线程，系统能够避免线程创建和销毁的开销；通过统一管理，系统能够更好地控制并发度和资源使用情况。

线程池的核心优势

性能优化

线程池通过复用现有线程显著减少了线程创建和销毁的开销。每次创建线程都需要分配内存、初始化线程栈等操作，而销毁线程同样需要回收资源。在高并发场景下，这些开销会严重影响系统性能。线程池通过预创建线程并重复使用，将这些开销降到最低。

public class PerformanceComparisonExample {private static final int TASK_COUNT = 1000;public static void main(String[] args) throws InterruptedException {// 测试传统线程创建性能long startTime = System.currentTimeMillis();testTraditionalThreads();long traditionalTime = System.currentTimeMillis() - startTime;// 测试线程池性能startTime = System.currentTimeMillis();testThreadPool();long threadPoolTime = System.currentTimeMillis() - startTime;System.out.println("传统线程创建耗时: " + traditionalTime + "ms");System.out.println("线程池执行耗时: " + threadPoolTime + "ms");System.out.println("性能提升: " + (traditionalTime - threadPoolTime) * 100 / traditionalTime + "%");}private static void testTraditionalThreads() throws InterruptedException {CountDownLatch latch = new CountDownLatch(TASK_COUNT);for (int i = 0; i < TASK_COUNT; i++) {new Thread(() -> {doWork();latch.countDown();}).start();}latch.await();}private static void testThreadPool() throws InterruptedException {ExecutorService executor = Executors.newFixedThreadPool(10);CountDownLatch latch = new CountDownLatch(TASK_COUNT);for (int i = 0; i < TASK_COUNT; i++) {executor.submit(() -> {doWork();latch.countDown();});}latch.await();executor.shutdown();}private static void doWork() {// 模拟实际工作负载for (int i = 0; i < 10000; i++) {Math.random();}}
}

资源控制

线程池能够精确控制系统中运行的线程数量，防止过多线程导致的资源耗尽问题。通过设置合理的核心线程数和最大线程数，可以确保系统在高负载情况下仍能稳定运行。

public class ResourceControlExample {public static void main(String[] args) {// 创建自定义线程池，严格控制线程数量ThreadPoolExecutor executor = new ThreadPoolExecutor(2,    // 核心线程数4,    // 最大线程数60L,  // 空闲线程存活时间TimeUnit.SECONDS,new ArrayBlockingQueue<>(10), // 有界队列，防止内存溢出new ThreadFactory() {private AtomicInteger threadNum = new AtomicInteger(1);@Overridepublic Thread newThread(Runnable r) {Thread t = new Thread(r, "CustomPool-" + threadNum.getAndIncrement());System.out.println("创建新线程: " + t.getName());return t;}},new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略);// 提交大量任务测试资源控制for (int i = 0; i < 20; i++) {final int taskId = i;try {executor.submit(() -> {System.out.println("任务 " + taskId + " 开始执行 - " + Thread.currentThread().getName());try {Thread.sleep(2000);} catch (InterruptedException e) {Thread.currentThread().interrupt();}System.out.println("任务 " + taskId + " 执行完成");});} catch (Exception e) {System.out.println("任务 " + taskId + " 提交失败: " + e.getMessage());}}// 监控线程池状态monitorThreadPool(executor);executor.shutdown();}private static void monitorThreadPool(ThreadPoolExecutor executor) {new Thread(() -> {while (!executor.isTerminated()) {System.out.printf("线程池状态 - 核心线程数: %d, 活跃线程数: %d, " +"队列大小: %d, 已完成任务数: %d%n",executor.getCorePoolSize(),executor.getActiveCount(),executor.getQueue().size(),executor.getCompletedTaskCount());try {Thread.sleep(1000);} catch (InterruptedException e) {Thread.currentThread().interrupt();break;}}}).start();}
}

响应速度提升

由于线程已经预先创建并处于等待状态，当任务到达时可以立即开始执行，无需等待线程创建过程。这种机制显著提升了系统的响应速度，特别是在处理大量短时间任务时效果更加明显。

统一管理

线程池提供了统一的线程管理接口，使得开发人员能够方便地监控线程状态、控制线程行为，并实现更加精细的线程调度策略。

Java线程池的实现架构

ThreadPoolExecutor类

ThreadPoolExecutor是Java线程池的核心实现类，它提供了完整的线程池功能。该类通过以下几个关键参数来控制线程池的行为：

public class ThreadPoolExecutorExample {public static void main(String[] args) {// 创建自定义ThreadPoolExecutorThreadPoolExecutor executor = new ThreadPoolExecutor(3,                                    // corePoolSize: 核心线程数6,                                    // maximumPoolSize: 最大线程数60L,                                  // keepAliveTime: 线程存活时间TimeUnit.SECONDS,                     // unit: 时间单位new LinkedBlockingQueue<>(100),       // workQueue: 工作队列new CustomThreadFactory(),            // threadFactory: 线程工厂new ThreadPoolExecutor.AbortPolicy()  // handler: 拒绝策略);// 提交任务演示各参数作用for (int i = 0; i < 20; i++) {final int taskId = i;executor.submit(new Task(taskId));}// 关闭线程池executor.shutdown();}static class Task implements Runnable {private final int taskId;public Task(int taskId) {this.taskId = taskId;}@Overridepublic void run() {System.out.println("任务 " + taskId + " 开始执行 - 线程: " + Thread.currentThread().getName());try {Thread.sleep(2000); // 模拟任务执行} catch (InterruptedException e) {Thread.currentThread().interrupt();}System.out.println("任务 " + taskId + " 执行完成");}}static class CustomThreadFactory implements ThreadFactory {private final AtomicInteger threadNumber = new AtomicInteger(1);@Overridepublic Thread newThread(Runnable r) {Thread t = new Thread(r, "CustomWorker-" + threadNumber.getAndIncrement());t.setDaemon(false);t.setPriority(Thread.NORM_PRIORITY);return t;}}
}

核心线程数（corePoolSize）定义了线程池中常驻线程的数量。这些线程在线程池启动后会一直存在，即使处于空闲状态也不会被回收。核心线程数的设置应该基于应用程序的基础负载来确定。

最大线程数（maximumPoolSize）限制了线程池中最多能够创建的线程数量。当任务队列满载且核心线程都在忙碌时，线程池会创建额外的线程来处理任务，但总数不会超过最大线程数。

线程存活时间（keepAliveTime）控制非核心线程的空闲存活时间。当这些线程空闲时间超过设定值时，会被自动回收以释放资源。

任务队列机制

线程池使用工作队列来缓存待执行的任务。当所有核心线程都在忙碌时，新提交的任务会被放入队列等待执行。Java提供了多种队列实现来满足不同的需求场景。

public class QueueTypesExample {public static void main(String[] args) throws InterruptedException {// 1. ArrayBlockingQueue - 有界队列demonstrateArrayBlockingQueue();Thread.sleep(1000);// 2. LinkedBlockingQueue - 链表队列demonstrateLinkedBlockingQueue();Thread.sleep(1000);// 3. SynchronousQueue - 同步队列demonstrateSynchronousQueue();Thread.sleep(1000);// 4. PriorityBlockingQueue - 优先级队列demonstratePriorityBlockingQueue();}private static void demonstrateArrayBlockingQueue() {System.out.println("=== ArrayBlockingQueue 演示 ===");ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 2, 0L, TimeUnit.MILLISECONDS,new ArrayBlockingQueue<>(3), // 容量为3的有界队列new ThreadPoolExecutor.CallerRunsPolicy());for (int i = 0; i < 8; i++) {final int taskId = i;executor.submit(() -> {System.out.println("ArrayBlockingQueue - 任务 " + taskId + " 执行中 - " + Thread.currentThread().getName());try {Thread.sleep(1000);} catch (InterruptedException e) {Thread.currentThread().interrupt();}});}executor.shutdown();}private static void demonstrateLinkedBlockingQueue() {System.out.println("=== LinkedBlockingQueue 演示 ===");ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 4, 60L, TimeUnit.SECONDS,new LinkedBlockingQueue<>(5) // 容量为5的链表队列);for (int i = 0; i < 10; i++) {final int taskId = i;executor.submit(() -> {System.out.println("LinkedBlockingQueue - 任务 " + taskId + " 执行中 - " + Thread.currentThread().getName());try {Thread.sleep(1000);} catch (InterruptedException e) {Thread.currentThread().interrupt();}});}executor.shutdown();}private static void demonstrateSynchronousQueue() {System.out.println("=== SynchronousQueue 演示 ===");ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 3, 60L, TimeUnit.SECONDS,new SynchronousQueue<>() // 不存储任务的同步队列);for (int i = 0; i < 5; i++) {final int taskId = i;executor.submit(() -> {System.out.println("SynchronousQueue - 任务 " + taskId + " 执行中 - " + Thread.currentThread().getName());try {Thread.sleep(1000);} catch (InterruptedException e) {Thread.currentThread().interrupt();}});}executor.shutdown();}private static void demonstratePriorityBlockingQueue() {System.out.println("=== PriorityBlockingQueue 演示 ===");ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 2, 0L, TimeUnit.MILLISECONDS,new PriorityBlockingQueue<>() // 优先级队列);// 提交不同优先级的任务executor.submit(new PriorityTask(3, "低优先级任务"));executor.submit(new PriorityTask(1, "高优先级任务"));executor.submit(new PriorityTask(2, "中优先级任务"));executor.submit(new PriorityTask(1, "高优先级任务2"));executor.shutdown();}static class PriorityTask implements Runnable, Comparable<PriorityTask> {private final int priority;private final String name;public PriorityTask(int priority, String name) {this.priority = priority;this.name = name;}@Overridepublic void run() {System.out.println("执行 " + name + " (优先级: " + priority + ") - " + Thread.currentThread().getName());try {Thread.sleep(1000);} catch (InterruptedException e) {Thread.currentThread().interrupt();}}@Overridepublic int compareTo(PriorityTask other) {return Integer.compare(this.priority, other.priority);}}
}

ArrayBlockingQueue是基于数组实现的有界阻塞队列，具有固定的容量限制。这种队列适合对内存使用有严格要求的场景，能够防止任务积压导致的内存溢出问题。

LinkedBlockingQueue是基于链表实现的阻塞队列，可以是有界的也可以是无界的。无界队列理论上可以存储无限数量的任务，但在实际应用中需要谨慎使用，避免内存耗尽。

SynchronousQueue是一种特殊的队列，它不存储任务，而是直接将任务从生产者传递给消费者。这种队列适合任务处理速度很快的场景。

PriorityBlockingQueue支持任务优先级排序，能够确保高优先级任务优先执行。这种队列适合需要任务调度优化的应用场景。

拒绝策略

当线程池无法接受新任务时（通常是因为队列已满且线程数已达上限），需要通过拒绝策略来处理这种情况。Java提供了四种内置的拒绝策略。

public class RejectionPolicyExample {public static void main(String[] args) throws InterruptedException {// 1. AbortPolicy - 抛出异常demonstrateAbortPolicy();Thread.sleep(2000);// 2. CallerRunsPolicy - 调用者执行demonstrateCallerRunsPolicy();Thread.sleep(2000);// 3. DiscardPolicy - 静默丢弃demonstrateDiscardPolicy();Thread.sleep(2000);// 4. DiscardOldestPolicy - 丢弃最老任务demonstrateDiscardOldestPolicy();Thread.sleep(2000);// 5. 自定义拒绝策略demonstrateCustomRejectionPolicy();}private static void demonstrateAbortPolicy() {System.out.println("=== AbortPolicy 演示 ===");ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,new ArrayBlockingQueue<>(1),new ThreadPoolExecutor.AbortPolicy());try {for (int i = 0; i < 5; i++) {final int taskId = i;executor.submit(() -> {System.out.println("AbortPolicy - 任务 " + taskId + " 执行中");try {Thread.sleep(2000);} catch (InterruptedException e) {Thread.currentThread().interrupt();}});}} catch (RejectedExecutionException e) {System.out.println("AbortPolicy - 任务被拒绝: " + e.getMessage());}executor.shutdown();}private static void demonstrateCallerRunsPolicy() {System.out.println("=== CallerRunsPolicy 演示 ===");ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,new ArrayBlockingQueue<>(1),new ThreadPoolExecutor.CallerRunsPolicy());for (int i = 0; i < 5; i++) {final int taskId = i;System.out.println("提交任务 " + taskId + " - 当前线程: " + Thread.currentThread().getName());executor.submit(() -> {System.out.println("CallerRunsPolicy - 任务 " + taskId + " 执行中 - " + Thread.currentThread().getName());try {Thread.sleep(1000);} catch (InterruptedException e) {Thread.currentThread().interrupt();}});}executor.shutdown();}private static void demonstrateDiscardPolicy() {System.out.println("=== DiscardPolicy 演示 ===");ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,new ArrayBlockingQueue<>(1),new ThreadPoolExecutor.DiscardPolicy());for (int i = 0; i < 5; i++) {final int taskId = i;executor.submit(() -> {System.out.println("DiscardPolicy - 任务 " + taskId + " 执行中");try {Thread.sleep(1000);} catch (InterruptedException e) {Thread.currentThread().interrupt();}});}System.out.println("DiscardPolicy - 部分任务将被静默丢弃");executor.shutdown();}private static void demonstrateDiscardOldestPolicy() {System.out.println("=== DiscardOldestPolicy 演示 ===");ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,new ArrayBlockingQueue<>(2),new ThreadPoolExecutor.DiscardOldestPolicy());for (int i = 0; i < 6; i++) {final int taskId = i;executor.submit(() -> {System.out.println("DiscardOldestPolicy - 任务 " + taskId + " 执行中");try {Thread.sleep(1000);} catch (InterruptedException e) {Thread.currentThread().interrupt();}});}executor.shutdown();}private static void demonstrateCustomRejectionPolicy() {System.out.println("=== 自定义拒绝策略演示 ===");ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,new ArrayBlockingQueue<>(1),new CustomRejectionHandler());for (int i = 0; i < 5; i++) {final int taskId = i;executor.submit(() -> {System.out.println("CustomPolicy - 任务 " + taskId + " 执行中");try {Thread.sleep(1000);} catch (InterruptedException e) {Thread.currentThread().interrupt();}});}executor.shutdown();}static class CustomRejectionHandler implements RejectedExecutionHandler {@Overridepublic void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {System.out.println("自定义拒绝策略: 任务被拒绝，记录日志并尝试重新提交");// 可以实现重试逻辑、日志记录、告警等try {Thread.sleep(100);if (!executor.isShutdown()) {executor.getQueue().offer(r);}} catch (InterruptedException e) {Thread.currentThread().interrupt();}}}
}

AbortPolicy是默认策略，它会抛出RejectedExecutionException异常，让调用者知道任务提交失败。这种策略适合需要明确处理任务拒绝情况的应用。

CallerRunsPolicy让调用线程直接执行被拒绝的任务，这样可以降低任务提交的速度，起到一定的背压作用。

DiscardPolicy静默丢弃被拒绝的任务，不进行任何处理。这种策略适合对任务丢失不敏感的场景。

DiscardOldestPolicy丢弃队列中最老的任务，为新任务腾出空间。这种策略适合希望优先处理新任务的场景。ortPolicy是默认策略，它会抛出RejectedExecutionException异常，让调用者知道任务提交失败。这种策略适合需要明确处理任务拒绝情况的应用。

CallerRunsPolicy让调用线程直接执行被拒绝的任务，这样可以降低任务提交的速度，起到一定的背压作用。

DiscardPolicy静默丢弃被拒绝的任务，不进行任何处理。这种策略适合对任务丢失不敏感的场景。

DiscardOldestPolicy丢弃队列中最老的任务，为新任务腾出空间。这种策略适合希望优先处理新任务的场景。

线程池的工作流程

线程池的任务执行遵循明确的优先级规则。当新任务提交到线程池时，首先检查核心线程是否都在忙碌。如果有空闲的核心线程，任务会立即分配给该线程执行。

public class ThreadPoolWorkflowExample {public static void main(String[] args) throws InterruptedException {// 创建线程池演示工作流程ThreadPoolExecutor executor = new ThreadPoolExecutor(2,    // 核心线程数4,    // 最大线程数5L,   // 非核心线程存活时间TimeUnit.SECONDS,new ArrayBlockingQueue<>(3), // 队列容量为3new WorkflowThreadFactory(),new ThreadPoolExecutor.AbortPolicy());// 演示任务提交和执行流程System.out.println("=== 线程池工作流程演示 ===");// 第一阶段：核心线程处理任务System.out.println("阶段1: 提交前2个任务，由核心线程处理");submitTask(executor, 1, "核心线程处理");submitTask(executor, 2, "核心线程处理");Thread.sleep(500);printPoolStatus(executor);// 第二阶段：任务进入队列System.out.println("\n阶段2: 提交3个任务，进入等待队列");submitTask(executor, 3, "进入队列等待");submitTask(executor, 4, "进入队列等待");submitTask(executor, 5, "进入队列等待");Thread.sleep(500);printPoolStatus(executor);// 第三阶段：创建非核心线程System.out.println("\n阶段3: 提交2个任务，创建非核心线程");submitTask(executor, 6, "触发非核心线程创建");submitTask(executor, 7, "触发非核心线程创建");Thread.sleep(500);printPoolStatus(executor);// 第四阶段：触发拒绝策略System.out.println("\n阶段4: 提交任务触发拒绝策略");try {submitTask(executor, 8, "将被拒绝");} catch (RejectedExecutionException e) {System.out.println("任务8被拒绝: " + e.getMessage());}// 等待任务执行完成Thread.sleep(8000);System.out.println("\n任务执行完成后的线程池状态:");printPoolStatus(executor);executor.shutdown();}private static void submitTask(ThreadPoolExecutor executor, int taskId, String description) {executor.submit(() -> {System.out.println("任务" + taskId + "开始执行 (" + description + ") - " + Thread.currentThread().getName());try {Thread.sleep(3000); // 模拟任务执行时间} catch (InterruptedException e) {Thread.currentThread().interrupt();}System.out.println("任务" + taskId + "执行完成 - " + Thread.currentThread().getName());});}private static void printPoolStatus(ThreadPoolExecutor executor) {System.out.printf("线程池状态 - 核心线程数: %d, 当前线程数: %d, " +"活跃线程数: %d, 队列大小: %d, 已完成任务数: %d%n",executor.getCorePoolSize(),executor.getPoolSize(),executor.getActiveCount(),executor.getQueue().size(),executor.getCompletedTaskCount());}static class WorkflowThreadFactory implements ThreadFactory {private final AtomicInteger threadNumber = new AtomicInteger(1);@Overridepublic Thread newThread(Runnable r) {Thread t = new Thread(r, "WorkflowThread-" + threadNumber.getAndIncrement());System.out.println("创建新线程: " + t.getName());return t;}}
}

如果所有核心线程都在忙碌，线程池会尝试将任务放入工作队列。如果队列未满，任务会在队列中等待空闲线程来处理。

当工作队列也已满时，线程池会考虑创建新的线程来处理任务，前提是当前线程总数未达到最大线程数限制。这些额外创建的线程称为非核心线程。

如果线程数已达到最大值且队列已满，新提交的任务将触发拒绝策略，根据配置的策略进行相应处理。

任务执行完成后，线程会返回线程池继续等待新的任务。非核心线程在空闲时间超过设定阈值后会被自动回收。

常见线程池类型

FixedThreadPool

固定大小线程池通过Executors.newFixedThreadPool()方法创建，它维护固定数量的线程。这种线程池的核心线程数和最大线程数相等，使用无界的LinkedBlockingQueue作为工作队列。固定线程池适合执行长期运行的任务，能够保证系统资源的稳定使用。

public class FixedThreadPoolExample {public static void main(String[] args) throws InterruptedException {// 创建固定大小的线程池ExecutorService fixedPool = Executors.newFixedThreadPool(3);System.out.println("=== FixedThreadPool 演示 ===");// 提交10个任务for (int i = 1; i <= 10; i++) {final int taskId = i;fixedPool.submit(() -> {System.out.println("FixedThreadPool - 任务" + taskId + "开始执行 - " + Thread.currentThread().getName());try {Thread.sleep(2000); // 模拟长时间任务} catch (InterruptedException e) {Thread.currentThread().interrupt();}System.out.println("FixedThreadPool - 任务" + taskId + "执行完成");});}// 等待所有任务完成fixedPool.shutdown();if (fixedPool.awaitTermination(30, TimeUnit.SECONDS)) {System.out.println("FixedThreadPool - 所有任务执行完成");} else {System.out.println("FixedThreadPool - 任务执行超时");fixedPool.shutdownNow();}}
}

CachedThreadPool

缓存线程池通过Executors.newCachedThreadPool()方法创建，它的特点是线程数量可以根据需要动态调整。该线程池的核心线程数为0，最大线程数为Integer.MAX_VALUE，使用SynchronousQueue作为工作队列。缓存线程池适合处理大量短时间的异步任务。

public class CachedThreadPoolExample {public static void main(String[] args) throws InterruptedException {// 创建缓存线程池ExecutorService cachedPool = Executors.newCachedThreadPool();System.out.println("=== CachedThreadPool 演示 ===");// 第一波任务：短时间大量任务System.out.println("提交第一波任务（短时间任务）:");for (int i = 1; i <= 20; i++) {final int taskId = i;cachedPool.submit(() -> {System.out.println("CachedThreadPool - 短任务" + taskId + " - " + Thread.currentThread().getName());try {Thread.sleep(100); // 短时间任务} catch (InterruptedException e) {Thread.currentThread().interrupt();}});}// 等待第一波任务完成Thread.sleep(2000);// 第二波任务：测试线程复用System.out.println("\n提交第二波任务（测试线程复用）:");for (int i = 21; i <= 30; i++) {final int taskId = i;cachedPool.submit(() -> {System.out.println("CachedThreadPool - 复用任务" + taskId + " - " + Thread.currentThread().getName());try {Thread.sleep(500);} catch (InterruptedException e) {Thread.currentThread().interrupt();}});}cachedPool.shutdown();if (cachedPool.awaitTermination(10, TimeUnit.SECONDS)) {System.out.println("CachedThreadPool - 所有任务执行完成");}}
}

SingleThreadExecutor

单线程执行器通过Executors.newSingleThreadExecutor()方法创建，它只使用一个工作线程来执行任务。这种线程池保证任务按照提交顺序依次执行，适合需要严格顺序控制的场景。

public class SingleThreadExecutorExample {public static void main(String[] args) throws InterruptedException {// 创建单线程执行器ExecutorService singlePool = Executors.newSingleThreadExecutor();System.out.println("=== SingleThreadExecutor 演示 ===");// 提交多个任务，观察顺序执行for (int i = 1; i <= 5; i++) {final int taskId = i;singlePool.submit(() -> {System.out.println("SingleThreadExecutor - 任务" + taskId + "开始执行 - " + Thread.currentThread().getName() + " - 时间: " + System.currentTimeMillis());try {Thread.sleep(1000);} catch (InterruptedException e) {Thread.currentThread().interrupt();}System.out.println("SingleThreadExecutor - 任务" + taskId + "执行完成");});}// 演示任务顺序性singlePool.submit(() -> {System.out.println("SingleThreadExecutor - 最后一个任务 - " + Thread.currentThread().getName());});singlePool.shutdown();if (singlePool.awaitTermination(10, TimeUnit.SECONDS)) {System.out.println("SingleThreadExecutor - 所有任务按顺序执行完成");}}
}

ScheduledThreadPool

定时线程池通过Executors.newScheduledThreadPool()方法创建，它支持任务的定时和周期性执行。这种线程池继承了ThreadPoolExecutor的所有功能，并添加了时间调度能力。

public class ScheduledThreadPoolExample {public static void main(String[] args) throws InterruptedException {// 创建定时线程池ScheduledExecutorService scheduledPool = Executors.newScheduledThreadPool(3);System.out.println("=== ScheduledThreadPool 演示 ===");System.out.println("当前时间: " + new Date());// 1. 延迟执行任务scheduledPool.schedule(() -> {System.out.println("延迟任务执行 - " + new Date() + " - " + Thread.currentThread().getName());}, 3, TimeUnit.SECONDS);// 2. 固定频率执行任务ScheduledFuture<?> fixedRateTask = scheduledPool.scheduleAtFixedRate(() -> {System.out.println("固定频率任务 - " + new Date() + " - " + Thread.currentThread().getName());try {Thread.sleep(1000); // 模拟任务执行时间} catch (InterruptedException e) {Thread.currentThread().interrupt();}}, 2, 3, TimeUnit.SECONDS); // 延迟2秒开始，每3秒执行一次// 3. 固定延迟执行任务ScheduledFuture<?> fixedDelayTask = scheduledPool.scheduleWithFixedDelay(() -> {System.out.println("固定延迟任务 - " + new Date() + " - " + Thread.currentThread().getName());try {Thread.sleep(2000); // 模拟任务执行时间} catch (InterruptedException e) {Thread.currentThread().interrupt();}}, 1, 4, TimeUnit.SECONDS); // 延迟1秒开始，任务完成后延迟4秒再执行// 4. 多个一次性延迟任务for (int i = 1; i <= 3; i++) {final int taskId = i;scheduledPool.schedule(() -> {System.out.println("一次性延迟任务" + taskId + " - " + new Date());}, i * 2, TimeUnit.SECONDS);}// 运行15秒后停止周期性任务Thread.sleep(15000);System.out.println("取消周期性任务...");fixedRateTask.cancel(false);fixedDelayTask.cancel(false);// 优雅关闭scheduledPool.shutdown();if (scheduledPool.awaitTermination(5, TimeUnit.SECONDS)) {System.out.println("ScheduledThreadPool 关闭完成");} else {scheduledPool.shutdownNow();}}
}

最佳实践与配置建议

线程数量配置

线程池大小的配置需要根据具体的应用场景来确定。对于CPU密集型任务，线程数量通常设置为CPU核心数加1，这样能够充分利用CPU资源而不会产生过多的上下文切换开销。

对于IO密集型任务，由于线程经常处于等待状态，可以设置更多的线程数量。一般建议设置为CPU核心数的2倍或更多，具体数值需要通过性能测试来确定最优值。

队列选择策略

工作队列的选择对线程池性能有重要影响。有界队列能够防止内存溢出，但可能导致任务被拒绝。无界队列避免了任务拒绝问题，但可能在高负载时消耗大量内存。

在实际应用中，建议使用有界队列并设置合理的容量。队列大小可以根据系统的内存容量和任务处理速度来确定。

监控和调优

线程池的运行状态需要持续监控以确保最优性能。重要的监控指标包括活跃线程数、队列长度、任务完成数量和拒绝任务数量。通过这些指标可以判断线程池配置是否合理。

当发现线程池性能不佳时，可以通过调整核心线程数、最大线程数、队列大小等参数来优化性能。调优过程需要结合实际的业务负载情况进行测试验证。

异常处理

线程池中的任务执行异常需要妥善处理。未捕获的异常可能导致工作线程终止，影响线程池的正常运行。建议在任务代码中添加适当的异常处理机制，或者通过ThreadFactory为线程设置UncaughtExceptionHandler。

优雅关闭

应用程序关闭时应该优雅地关闭线程池，确保正在执行的任务能够完成。可以通过调用shutdown()方法来停止接收新任务，然后使用awaitTermination()方法等待现有任务完成。如果需要强制关闭，可以使用shutdownNow()方法。

性能优化技巧

预热策略

线程池启动时可以通过prestartAllCoreThreads()方法预先创建所有核心线程，避免首次任务提交时的线程创建延迟。这种预热策略特别适合对响应时间要求严格的应用场景。

任务分解

将大任务分解为多个小任务可以提高线程池的并行处理能力。合理的任务粒度既能充分利用多线程优势，又不会产生过多的管理开销。

线程本地变量

合理使用ThreadLocal可以避免线程间的数据竞争，提高并发性能。但需要注意ThreadLocal的内存泄漏问题，确保在适当时机清理数据。

常见问题与解决方案

内存泄漏

线程池可能因为任务对象持有大量引用而导致内存泄漏。解决方案包括及时清理任务中的引用、使用弱引用、合理设置队列大小等。

死锁问题

当线程池中的任务相互等待时可能发生死锁。避免死锁的方法包括避免嵌套锁、使用超时机制、合理设计任务依赖关系等。

性能瓶颈

线程池性能瓶颈可能出现在任务提交、队列操作、线程调度等环节。通过性能分析工具定位瓶颈点，然后采用相应的优化策略。

http://www.xdnf.cn/news/12830.html

相关文章：

  • nginx配置文件
  • leetcode238-除自身以外数组的乘积
  • 【JVM面试篇】高频八股汇总——Java内存区域
  • 华为OD机考 - 水仙花数 Ⅰ（2025B卷 100分）
  • 8. 二叉树（随想录）
  • 本地缓存在Java中的实现方式
  • “图像说话，文本有图”——用Python玩转跨模态数据关联分析
  • 【2025CVPR】模型融合新范式：PLeaS算法详解（基于排列与最小二乘的模型合并技术）
  • 飞云控盘指标-副图指标-买点一持仓操作技术图文解说
  • 初级程序员入门指南
  • 跟进一下目前最新的大数据技术
  • 设备驱动与文件系统：06 目录与文件
  • 骨盆-x光参数
  • python生成器
  • SWAN(Scade One) 语言原理介绍
  • Linux中《进程控制》详细介绍
  • RootSIFT的目标定位，opencvsharp。
  • DOM（文档对象模型）深度解析
  • 开源项目实战学习之YOLO11：12.6 ultralytics-models-tiny_encoder.py
  • 【深度学习-Day 25】告别过拟合：深入解析 L1 与 L2 正则化（权重衰减）的原理与实战
  • 标准代码项目开发流程学习指南
  • CMS内容管理系统的设计与实现：架构设计
  • 红黑树完全指南：为何工程都用它？原理、实现、场景、误区全解析
  • 数学：”度量空间”了解一下？
  • JESD204B IP核接口实例，ADI的ADRV9009板卡，ZYNQ7045驱动实现2发2收。
  • LLMs 系列科普文（14）
  • 关于IE浏览器被绑定安装，还卸载不掉
  • 72常用控件_QGridLayout的使用
  • 热成像实例分割电力设备数据集（3类，838张）
  • Vscode下Go语言环境配置