一、为什么要使用线程池？

在深入ThreadPoolExecutor之前，我们先明确使用线程池的必要性：

  • 降低资源消耗：通过重复利用已创建的线程，降低线程创建和销毁造成的消耗
  • 提高响应速度：当任务到达时，任务可以不需要等待线程创建就能立即执行
  • 提高线程可管理性：线程是稀缺资源，使用线程池可以进行统一分配、调优和监控
  • 提供更强大的功能：线程池具备可扩展性，允许开发人员向其中增加更多的功能

二、ThreadPoolExecutor类结构概览

public class ThreadPoolExecutor extends AbstractExecutorService {
    // 核心字段
    private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
    private final BlockingQueue<Runnable> workQueue;
    private final HashSet<Worker> workers = new HashSet<Worker>();
    private volatile ThreadFactory threadFactory;
    private volatile RejectedExecutionHandler handler;
    private volatile long keepAliveTime;
    private volatile int corePoolSize;
    private volatile int maximumPoolSize;
    
    // 主要构造方法
    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {
        // 参数校验和初始化
    }
}

三、七大核心参数深度解析

3.1 corePoolSize - 核心线程数

作用：线程池中保持存活的最小线程数量，即使这些线程处于空闲状态。

工作机制：

  • 当提交新任务时，如果当前线程数 < corePoolSize，即使存在空闲线程，也会创建新线程
  • 默认情况下，核心线程会一直存活，除非设置allowCoreThreadTimeOut
public class CorePoolSizeExample {
    public static void main(String[] args) {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
            2,  // corePoolSize - 核心线程数
            5,  // maximumPoolSize
            60, // keepAliveTime
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(10)
        );
        
        // 提交任务
        for (int i = 0; i < 3; i++) {
            final int taskId = i;
            executor.execute(() -> {
                System.out.println("Task " + taskId + " executed by " + 
                    Thread.currentThread().getName());
            });
        }
        
        // 输出线程池状态
        System.out.println("核心线程数: " + executor.getCorePoolSize());
        System.out.println("当前线程数: " + executor.getPoolSize());
        
        executor.shutdown();
    }
}

配置建议：

  • CPU密集型任务：CPU核心数 + 1
  • IO密集型任务：CPU核心数 × 2
  • 混合型任务：(CPU核心数 × IO等待时间) / CPU计算时间

3.2 maximumPoolSize - 最大线程数

作用：线程池允许创建的最大线程数量。

工作机制：

  • 当workQueue已满且当前线程数 < maximumPoolSize时，会创建新线程
  • 达到maximumPoolSize后，新任务会触发拒绝策略
public class MaximumPoolSizeExample {
    public static void main(String[] args) throws InterruptedException {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
            2,  // corePoolSize
            4,  // maximumPoolSize - 最大线程数
            1,  // keepAliveTime
            TimeUnit.SECONDS,
            new SynchronousQueue<>() // 无缓冲队列
        );
        
        // 提交6个任务，观察线程创建情况
        for (int i = 0; i < 6; i++) {
            final int taskId = i;
            try {
                executor.execute(() -> {
                    try {
                        Thread.sleep(1000);
                        System.out.println("Task " + taskId + " completed by " + 
                            Thread.currentThread().getName());
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                });
                System.out.println("Task " + taskId + " submitted");
            } catch (RejectedExecutionException e) {
                System.out.println("Task " + taskId + " rejected");
            }
        }
        
        Thread.sleep(3000);
        System.out.println("最终线程数: " + executor.getPoolSize());
        executor.shutdown();
    }
}

3.3 keepAliveTime - 线程存活时间

作用：当线程数大于核心线程数时，多余的空闲线程在终止前等待新任务的最长时间。

工作机制：

  • 仅对超出corePoolSize的线程有效
  • 配合TimeUnit单位使用
  • 设置allowCoreThreadTimeOut为true时，核心线程也会超时终止
public class KeepAliveTimeExample {
    public static void main(String[] args) throws InterruptedException {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
            2,  // corePoolSize
            4,  // maximumPoolSize
            2,  // keepAliveTime - 线程存活时间
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(2)
        );
        
        // 允许核心线程超时
        executor.allowCoreThreadTimeOut(true);
        
        System.out.println("初始线程数: " + executor.getPoolSize());
        
        // 提交一批任务
        for (int i = 0; i < 6; i++) {
            final int taskId = i;
            executor.execute(() -> {
                System.out.println("Task " + taskId + " started by " + 
                    Thread.currentThread().getName());
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
        }
        
        Thread.sleep(1000);
        System.out.println("任务执行后线程数: " + executor.getPoolSize());
        
        Thread.sleep(3000); // 等待超时
        System.out.println("超时后线程数: " + executor.getPoolSize());
        
        executor.shutdown();
    }
}

3.4 TimeUnit - 时间单位

作用：keepAliveTime参数的时间单位。

常用单位：

  • TimeUnit.NANOSECONDS：纳秒
  • TimeUnit.MICROSECONDS：微秒
  • TimeUnit.MILLISECONDS：毫秒
  • TimeUnit.SECONDS：秒
  • TimeUnit.MINUTES：分钟
  • TimeUnit.HOURS：小时
  • TimeUnit.DAYS：天

3.5 workQueue - 工作队列

作用：用于保存等待执行的任务的阻塞队列。

常用队列类型及适用场景：

3.5.1 SynchronousQueue - 直接传递队列
public class SynchronousQueueExample {
    public static void main(String[] args) {
        // SynchronousQueue：不存储元素，每个插入操作必须等待另一个线程的移除操作
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
            2, 5, 60, TimeUnit.SECONDS, 
            new SynchronousQueue<>()
        );
        
        // 适用于任务处理速度快的场景，能够快速创建新线程
    }
}
3.5.2 LinkedBlockingQueue - 无界队列
public class LinkedBlockingQueueExample {
    public static void main(String[] args) {
        // LinkedBlockingQueue：基于链表的无界队列（默认Integer.MAX_VALUE）
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
            2, 5, 60, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>() // 无界队列，maximumPoolSize无效
        );
        
        // 适用于任务量不可预测但需要平滑处理的场景
    }
}
3.5.3 ArrayBlockingQueue - 有界队列
public class ArrayBlockingQueueExample {
    public static void main(String[] args) {
        // ArrayBlockingQueue：基于数组的有界队列
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
            2, 5, 60, TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(10) // 有界队列，容量10
        );
        
        // 适用于需要控制资源使用的场景
    }
}
3.5.4 PriorityBlockingQueue - 优先级队列
public class PriorityBlockingQueueExample {
    static class PriorityTask implements Runnable, Comparable<PriorityTask> {
        private int priority;
        private String name;
        
        public PriorityTask(String name, int priority) {
            this.name = name;
            this.priority = priority;
        }
        
        @Override
        public void run() {
            System.out.println("Executing: " + name + " with priority: " + priority);
        }
        
        @Override
        public int compareTo(PriorityTask other) {
            return Integer.compare(other.priority, this.priority); // 降序
        }
    }
    
    public static void main(String[] args) {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
            2, 5, 60, TimeUnit.SECONDS,
            new PriorityBlockingQueue<>()
        );
        
        // 提交不同优先级的任务
        executor.execute(new PriorityTask("High", 10));
        executor.execute(new PriorityTask("Low", 1));
        executor.execute(new PriorityTask("Medium", 5));
        
        executor.shutdown();
    }
}

3.6 ThreadFactory - 线程工厂

作用：用于创建新线程，可以定制线程的属性。

public class CustomThreadFactoryExample {
    
    static class CustomThreadFactory implements ThreadFactory {
        private final AtomicInteger threadNumber = new AtomicInteger(1);
        private final String namePrefix;
        private final ThreadGroup group;
        
        public CustomThreadFactory(String poolName) {
            SecurityManager s = System.getSecurityManager();
            group = (s != null) ? s.getThreadGroup() : 
                              Thread.currentThread().getThreadGroup();
            namePrefix = poolName + "-thread-";
        }
        
        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(group, r, 
                                namePrefix + threadNumber.getAndIncrement(),
                                0);
            // 设置线程属性
            if (t.isDaemon())
                t.setDaemon(false);
            if (t.getPriority() != Thread.NORM_PRIORITY)
                t.setPriority(Thread.NORM_PRIORITY);
            
            // 设置异常处理器
            t.setUncaughtExceptionHandler((thread, throwable) -> {
                System.err.println("Uncaught exception in thread: " + thread.getName());
                throwable.printStackTrace();
            });
            
            return t;
        }
    }
    
    public static void main(String[] args) {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
            2, 5, 60, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(10),
            new CustomThreadFactory("CustomPool")  // 自定义线程工厂
        );
        
        executor.execute(() -> System.out.println(
            "Task executed by: " + Thread.currentThread().getName()));
        
        executor.shutdown();
    }
}

3.7 RejectedExecutionHandler - 拒绝策略

作用：当线程池和队列都已满时，处理新提交的任务。

3.7.1 内置拒绝策略
public class RejectionPolicyExample {
    
    public static void main(String[] args) {
        // 1. AbortPolicy - 默认策略，抛出RejectedExecutionException
        testRejectionPolicy(new ThreadPoolExecutor.AbortPolicy(), "AbortPolicy");
        
        // 2. CallerRunsPolicy - 由调用者线程执行任务
        testRejectionPolicy(new ThreadPoolExecutor.CallerRunsPolicy(), "CallerRunsPolicy");
        
        // 3. DiscardPolicy - 静默丢弃任务
        testRejectionPolicy(new ThreadPoolExecutor.DiscardPolicy(), "DiscardPolicy");
        
        // 4. DiscardOldestPolicy - 丢弃队列中最老的任务
        testRejectionPolicy(new ThreadPoolExecutor.DiscardOldestPolicy(), "DiscardOldestPolicy");
    }
    
    static void testRejectionPolicy(RejectedExecutionHandler handler, String policyName) {
        System.out.println("\n=== Testing " + policyName + " ===");
        
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
            1, 1, 0, TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(1),
            handler
        );
        
        try {
            // 提交3个任务到容量为2的线程池（1个执行 + 1个队列）
            for (int i = 0; i < 3; i++) {
                final int taskId = i;
                try {
                    executor.execute(() -> {
                        try {
                            Thread.sleep(1000);
                            System.out.println("Task " + taskId + " completed");
                        } catch (InterruptedException e) {
                            Thread.currentThread().interrupt();
                        }
                    });
                    System.out.println("Task " + taskId + " submitted successfully");
                } catch (RejectedExecutionException e) {
                    System.out.println("Task " + taskId + " rejected: " + e.getMessage());
                }
            }
        } finally {
            executor.shutdown();
        }
    }
}
3.7.2 自定义拒绝策略
public class CustomRejectionPolicy {
    
    static class LoggingRejectionHandler implements RejectedExecutionHandler {
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            // 记录日志
            System.err.println("Task rejected: " + r.toString());
            System.err.println("Executor state: " + executor.toString());
            
            // 可选的降级策略
            if (!executor.isShutdown()) {
                System.out.println("Executing in caller thread as fallback");
                r.run(); // 在调用者线程中执行
            }
        }
    }
    
    static class RetryRejectionHandler implements RejectedExecutionHandler {
        private final int maxRetries;
        private final long retryDelayMs;
        
        public RetryRejectionHandler(int maxRetries, long retryDelayMs) {
            this.maxRetries = maxRetries;
            this.retryDelayMs = retryDelayMs;
        }
        
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            for (int retry = 0; retry < maxRetries; retry++) {
                try {
                    if (executor.getQueue().offer(r, retryDelayMs, TimeUnit.MILLISECONDS)) {
                        System.out.println("Task requeued after " + retry + " retries");
                        return;
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
            // 最终失败处理
            System.err.println("Task failed after " + maxRetries + " retries");
        }
    }
}

四、ThreadPoolExecutor执行流程深度剖析

4.1 核心执行流程

// 简化版的执行流程（实际源码在ThreadPoolExecutor.execute方法）
public void execute(Runnable command) {
    if (command == null) throw new NullPointerException();
    
    int c = ctl.get();
    
    // 阶段1：当前线程数 < corePoolSize
    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.2 完整执行流程图
是
否
是
否
是
否
线程池关闭
无工作线程
正常
提交任务
线程数 < corePoolSize?
创建核心线程执行任务
任务入队成功?
任务进入等待队列
线程数 < maximumPoolSize?
创建非核心线程执行任务
执行拒绝策略
需要重新检查?
移除任务并拒绝
创建新线程
等待执行

五、四种常用线程池的底层实现

5.1 通过Executors工具类创建

public class ExecutorsAnalysis {
    
    public static void main(String[] args) {
        // 1. FixedThreadPool - 固定线程数
        ExecutorService fixedPool = Executors.newFixedThreadPool(5);
        // 底层：new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, 
        //                              new LinkedBlockingQueue<Runnable>())
        
        // 2. CachedThreadPool - 可缓存线程池
        ExecutorService cachedPool = Executors.newCachedThreadPool();
        // 底层：new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS,
        //                              new SynchronousQueue<Runnable>())
        
        // 3. SingleThreadExecutor - 单线程池
        ExecutorService singlePool = Executors.newSingleThreadExecutor();
        // 底层：new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,
        //                              new LinkedBlockingQueue<Runnable>())
        
        // 4. ScheduledThreadPool - 定时任务线程池
        ScheduledExecutorService scheduledPool = Executors.newScheduledThreadPool(3);
        // 底层：new ScheduledThreadPoolExecutor(corePoolSize)
    }
}

六、线程池监控与调优实战

6.1 线程池状态监控

public class ThreadPoolMonitor {
    private final ThreadPoolExecutor executor;
    private final ScheduledExecutorService monitor;
    
    public ThreadPoolMonitor(ThreadPoolExecutor executor) {
        this.executor = executor;
        this.monitor = Executors.newSingleThreadScheduledExecutor();
    }
    
    public void startMonitoring() {
        monitor.scheduleAtFixedRate(this::collectMetrics, 0, 1, TimeUnit.SECONDS);
    }
    
    private void collectMetrics() {
        // 基础指标
        int poolSize = executor.getPoolSize();
        int activeCount = executor.getActiveCount();
        long completedTaskCount = executor.getCompletedTaskCount();
        long taskCount = executor.getTaskCount();
        int queueSize = executor.getQueue().size();
        
        // 计算饱和度
        double saturation = (double) activeCount / executor.getMaximumPoolSize();
        
        System.out.printf(
            "Pool: %d/%d, Active: %d, Queue: %d, Completed: %d, Saturation: %.2f%n",
            poolSize, executor.getMaximumPoolSize(), activeCount, 
            queueSize, completedTaskCount, saturation
        );
        
        // 预警机制
        if (saturation > 0.8) {
            System.err.println("WARNING: Thread pool saturation high: " + saturation);
        }
    }
    
    public void stopMonitoring() {
        monitor.shutdown();
    }
    
    // 动态调整线程池参数
    public void dynamicAdjust(int newCoreSize, int newMaxSize) {
        executor.setCorePoolSize(newCoreSize);
        executor.setMaximumPoolSize(newMaxSize);
        System.out.println("Thread pool adjusted: core=" + newCoreSize + ", max=" + newMaxSize);
    }
}

6.2 使用示例

public class ThreadPoolBestPractice {
    
    public static void main(String[] args) throws Exception {
        // 创建可监控的线程池
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
            2,  // corePoolSize
            10, // maximumPoolSize  
            60, // keepAliveTime
            TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(100),
            new CustomThreadFactory("BusinessPool"),
            new LoggingRejectionHandler()
        );
        
        // 启动监控
        ThreadPoolMonitor monitor = new ThreadPoolMonitor(executor);
        monitor.startMonitoring();
        
        // 模拟任务提交
        for (int i = 0; i < 200; i++) {
            final int taskId = i;
            executor.execute(() -> {
                try {
                    // 模拟业务处理
                    Thread.sleep(100 + (int)(Math.random() * 400));
                    System.out.println("Processed task: " + taskId);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
            
            // 控制提交速度
            if (i % 50 == 0) {
                Thread.sleep(1000);
            }
        }
        
        // 动态调整（基于监控数据）
        Thread.sleep(5000);
        monitor.dynamicAdjust(5, 15);
        
        // 优雅关闭
        executor.shutdown();
        if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
            executor.shutdownNow();
        }
        
        monitor.stopMonitoring();
    }
}

七、常见问题与解决方案

7.1 线程池配置最佳实践

public class ThreadPoolConfigGuide {
    
    /**
     * 根据业务类型推荐线程池配置
     */
    public static ThreadPoolExecutor createOptimalPool(WorkloadType type) {
        int cpuCores = Runtime.getRuntime().availableProcessors();
        
        switch (type) {
            case CPU_INTENSIVE:
                // CPU密集型：线程数不宜过多，避免频繁上下文切换
                return new ThreadPoolExecutor(
                    cpuCores, cpuCores + 1, 30L, TimeUnit.SECONDS,
                    new LinkedBlockingQueue<>(1000)
                );
                
            case IO_INTENSIVE:
                // IO密集型：可以配置更多线程
                return new ThreadPoolExecutor(
                    cpuCores * 2, cpuCores * 4, 60L, TimeUnit.SECONDS,
                    new LinkedBlockingQueue<>(2000)
                );
                
            case MIXED:
                // 混合型：适中配置
                return new ThreadPoolExecutor(
                    cpuCores, cpuCores * 2, 45L, TimeUnit.SECONDS,
                    new ArrayBlockingQueue<>(500)
                );
                
            default:
                throw new IllegalArgumentException("Unknown workload type");
        }
    }
    
    enum WorkloadType {
        CPU_INTENSIVE,  // 计算密集：加密、算法等
        IO_INTENSIVE,   // IO密集：网络请求、数据库操作
        MIXED           // 混合型
    }
}

7.2 避免的坑

  1. 无界队列导致内存溢出

    // 错误示例
new ThreadPoolExecutor(n, n, 0L, TimeUnit.MILLISECONDS, 
                      new LinkedBlockingQueue<>()); // 可能导致OOM

// 正确做法
new ThreadPoolExecutor(n, n, 0L, TimeUnit.MILLISECONDS,
                      new LinkedBlockingQueue<>(1000)); // 设置合理边界

  2. 不合理的线程超时设置

    // 错误：核心线程频繁创建销毁
executor.allowCoreThreadTimeOut(true);
executor.setKeepAliveTime(1, TimeUnit.SECONDS);

// 正确：根据业务特点设置合理超时
executor.setKeepAliveTime(60, TimeUnit.SECONDS); // 适度超时

八、总结

ThreadPoolExecutor是Java并发编程的核心组件，合理使用需要深入理解其七大参数：

  1. corePoolSize：根据任务类型合理设置，避免过度创建或资源浪费
  2. maximumPoolSize：设置合理的上限，防止资源耗尽
  3. keepAliveTime：平衡资源利用和响应速度
  4. workQueue：根据任务特点和系统负载选择合适的队列
  5. ThreadFactory：定制线程创建，便于监控和问题排查
  6. RejectedExecutionHandler：提供合适的降级策略

最佳实践建议：

  • 监控线程池运行状态，动态调整参数
  • 为不同业务场景创建独立的线程池
  • 设置合理的队列边界和拒绝策略
  • 在Spring等框架中合理配置线程池

通过深入理解ThreadPoolExecutor的工作原理和合理配置，可以构建出高性能、高可用的并发处理系统。

# java
Logo
全球具身智能开发者社区

立足具身智能前沿赛道，致力于搭建全球化、开源化、全栈式技术交流与实践共创平台。

更多推荐

cover

开源聚势·具身启智，杭州这场沙龙给出中国具身智能产业化新答案

avatar 全球具身智能开发者社区

YoMo安全机制详解：TLS v1.3如何保护你的AI代理通信

在当今AI驱动的分布式系统中，安全通信已成为不可忽视的核心需求。YoMo作为Serverless AI Agent Framework，采用TLS v1.3加密协议构建了强大的安全防护机制，确保AI代理在地理分布式边缘计算环境中的通信安全。本文将深入解析YoMo的TLS实现原理、配置方法及最佳实践，帮助开发者构建安全可靠的AI应用。## 为什么TLS v1.3是AI代理通信的理想选择TLS

avatar 全球具身智能开发者社区

InternScenes开源数据集

数据集介绍 InternScenes 是上海人工智能实验室发布的大规模、可模拟室内场景数据集，论文收录于 NeurIPS 2025。具身人工智能的发展高度依赖于具有场景多样性和逼真布局的大规模、可模拟3D场景数据集。然而，现有数据集通常存在以下不足：数据规模或多样性有限、布局经过"净化"处理导致小物体缺失，以及严重的物体碰撞问题。 为解决上述问题，InternScenes 整合

avatar 全球具身智能开发者社区