前面的文章中说了线程，开发者可以开启多个线程去完成多个任务，但是其实很多公司并不推荐单独开启线程，而是推荐使用线程池。

究其原因，一般是考虑线程的管理问题。线程虽然增加了处理任务的速度，但是如果使用不当，可能会造成内存泄露、程序崩溃等严重问题。所以了解线程池显得颇为重要。

基本概念

线程池可以望文生义理解为线程的池子，内部容纳了一定数量的线程，而且它还包含了线程的管理功能。Java中创建线程池的方法是：

public ThreadPoolExecutor(
    int corePoolSize,         // 核心线程数
    int maximumPoolSize,      // 最大线程数
    long keepAliveTime,       // 非核心线程存活时间
    TimeUnit unit,            // 存活时间单位
    BlockingQueue<Runnable> workQueue,  // 任务队列
    ThreadFactory threadFactory,        // 线程工厂
    RejectedExecutionHandler handler    // 拒绝策略
)

可以看到构造函数中包含了多个参数：

  • corePoolSize：核心线程数线程池长期维持的线程数量（即使线程空闲）。任务提交时，若当前线程数小于corePoolSize，线程池会优先创建新线程（核心线程）处理任务，而非放入队列。特例：若调用allowCoreThreadTimeOut(true)，核心线程也会在空闲超过keepAliveTime后销毁。
  • maximumPoolSize：最大线程数线程池允许创建的最大线程数（核心线程 + 非核心线程）。当任务队列满且核心线程都在工作时，线程池会创建非核心线程处理任务，直到线程数达到此值。

线程池中虽然有核心线程数的概念，但是其实并没有严格意义上区分核心线程和非核心线程，但是可以理解为：线程数<=核心线程数时，创建的都是核心线程；线程池执行完毕后，线程池缩减线程完成后，剩下的都是核心线程；如果allowCoreThreadTimeOut(true)，第一批消亡的线程都是非核心线程，第二批消亡的是核心线程数（但此时所有线程都会消亡，根本就没有必要区分核心非核心了）。
.
纯属个人见解~

  • keepAliveTime + unit：非核心线程存活时间当线程数超过corePoolSize时，空闲的非核心线程会在指定时间后被销毁，释放资源。合理设置此值可在 “突发任务” 后自动收缩线程数，避免资源浪费。
  • workQueue：任务队列用于缓冲等待执行的任务，仅当核心线程都在工作时才会使用。常见的队列类型有：
    • LinkedBlockingQueue：无界队列（默认容量Integer.MAX_VALUE），可能导致任务无限堆积引发 OOM；
    • ArrayBlockingQueue：有界队列，需指定容量，能控制任务积压上限；
    • SynchronousQueue：同步队列，不存储任务，需立即有线程接收，适合任务处理速度快的场景。
  • threadFactory：线程工厂用于创建线程的工厂，可自定义线程名称（便于日志排查）、优先级、是否为守护线程等。默认使用Executors.defaultThreadFactory()，创建的线程均为非守护线程，名称格式为pool-{poolNum}-thread-{threadNum}。
  • handler：拒绝策略当任务队列满且线程数达到maximumPoolSize时，新提交的任务会触发拒绝策略。JDK 默认提供 4 种策略：
    • AbortPolicy（默认）：直接抛出RejectedExecutionException，中断任务提交；
    • CallerRunsPolicy：让提交任务的线程（调用者）自己执行任务，减缓提交速度；
    • DiscardPolicy：默默丢弃新任务，不做任何处理；
    • DiscardOldestPolicy：丢弃队列中最旧的任务，再尝试提交新任务。

线程池Demo

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class ThreadPoolDemo {
  public static void main(String[] args) {
    ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 3, 10, TimeUnit.SECONDS, new ArrayBlockingQueue<>(3));
    for (int i = 0; i < 20; i++) {
      executor.execute(new Thread(()-> {
        System.out.println(Thread.currentThread().getName() + " 执行任务中");
      }));
    }
  }
}

输出:

pool-1-thread-1 执行任务中
pool-1-thread-3 执行任务中
pool-1-thread-2 执行任务中
pool-1-thread-1 执行任务中
pool-1-thread-3 执行任务中
Exception in thread "main" pool-1-thread-2 执行任务中
java.util.concurrent.RejectedExecutionException: Task Thread[Thread-6,5,main] rejected from java.util.concurrent.ThreadPoolExecutor@448139f0[Running, pool size = 3, active threads = 2, queued tasks = 2, completed tasks = 1]
	at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2063)
	at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:830)
	at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1379)
	at ThreadPoolDemo.main(ThreadPoolDemo.java:9)

从输出中可以看到，定义的循环20次任务，但实际上只执行了五次就报错了：java.util.concurrent.RejectedExecutionException，很明显是触发了线程池的拒绝策略。
但同时可以看出，线程最多的时候只创建了三个线程。

为了追溯这些问题，我们捞出线程池的构造方法：


    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {
        if (corePoolSize < 0 || // 核心线程数不能小于0
            maximumPoolSize <= 0 || // 最大线程数不能小于0
            maximumPoolSize < corePoolSize || // 核心线程数必须小于最大线程数
            keepAliveTime < 0) // 存活时间不能小于0
            throw new IllegalArgumentException();
        if (workQueue == null || threadFactory == null || handler == null)
            throw new NullPointerException(); // 任务队列不能为null，线程工厂不能为null，拒绝策略不能为null（可以不传（内部会使用默认值），但是不能传null）
        this.acc = System.getSecurityManager() == null ?
                null :
                AccessController.getContext();
        // 将外部传入的数据赋值
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }

所以此处触发拒绝策略的原因，是我们没有指定拒绝策略，因而线程池采用了默认的策略——拒绝策略

    private static final RejectedExecutionHandler defaultHandler =
        new AbortPolicy();

而我们定义了最大线程数为3，就限制了线程池最多创建的线程数量。

线程池执行任务的原理

    public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        /*
         * Proceed in 3 steps:
         *
         * 1. If fewer than corePoolSize threads are running, try to
         * start a new thread with the given command as its first
         * task.  The call to addWorker atomically checks runState and
         * workerCount, and so prevents false alarms that would add
         * threads when it shouldn't, by returning false.
         *
         * 2. If a task can be successfully queued, then we still need
         * to double-check whether we should have added a thread
         * (because existing ones died since last checking) or that
         * the pool shut down since entry into this method. So we
         * recheck state and if necessary roll back the enqueuing if
         * stopped, or start a new thread if there are none.
         *
         * 3. If we cannot queue task, then we try to add a new
         * thread.  If it fails, we know we are shut down or saturated
         * and so reject the task.
         */
        int c = ctl.get(); // ctl是一个AtomicInteger,前面3位代表线程池状态，后面29位代表线程数量
        if (workerCountOf(c) < corePoolSize) { // 如果线程数小于核心线程数，则创建核心线程数
            if (addWorker(command, true))
                return;
            c = ctl.get(); // 增加核心线程后，重新获取线程数（相当于刷新该变量）
        }
       // 1. 先检查线程池是否在运行，且任务能否成功加入阻塞队列
		if (isRunning(c) && workQueue.offer(command)) {
		    // 2. 任务入队后，二次检查线程池状态（防止入队后线程池突然关闭）
		    int recheck = ctl.get();
		    // 2.1 如果线程池已不运行，且成功从队列中移除了任务，就拒绝该任务
		    if (!isRunning(recheck) && remove(command))
		        reject(command);
		    // 2.2 如果线程池仍运行，且当前没有工作线程，就创建一个空任务线程（保证有线程处理队列任务）
		    else if (workerCountOf(recheck) == 0)
		        addWorker(null, false);
		}
		// 3. 如果任务入队失败（队列满），尝试创建新线程执行任务（非核心线程，因core参数为false）
		else if (!addWorker(command, false))
		    // 3.1 如果创建线程失败（如线程数已达最大值），则拒绝任务
		    reject(command);
    }

addWorker方法源代码：

addWorker 方法在 ThreadPoolExecutor 类中，主要用于在线程池中添加新的工作线程来执行任务。

private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    for (;;) {
        // 获取当前线程池状态和工作线程数量
        int c = ctl.get();
        int rs = runStateOf(c);

        // 检查线程池状态，如果线程池停止或者某些不允许添加线程的状态，返回false
        if (rs >= SHUTDOWN &&
           ! (rs == SHUTDOWN && firstTask == null &&! workQueue.isEmpty()))
            return false;

        for (;;) {
            // 获取当前工作线程数量
            int wc = workerCountOf(c);
            // 判断工作线程数量是否超过最大限制或根据传入的core参数判断是否超过核心线程数或最大线程数
            if (wc >= CAPACITY ||
                wc >= (core? corePoolSize : maximumPoolSize))
                return false;
            // 尝试原子性地增加工作线程数量，如果成功则跳出外层循环
            if (compareAndIncrementWorkerCount(c))
                break retry;
            // 如果增加失败，说明有其他线程也在操作，重新获取线程池状态
            c = ctl.get();
            if (runStateOf(c)!= rs)
                continue retry;
        }
    }

    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        // 创建Worker对象，传入第一个任务
        w = new Worker(firstTask);
        // 获取Worker中的线程
        final Thread t = w.thread;
        if (t!= null) {
            // 加锁，保证线程安全
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                // 再次检查线程池状态，确保添加线程是安全的
                int rs = runStateOf(ctl.get());
                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) {
                    if (t.isAlive()) 
                        throw new IllegalThreadStateException();
                    // 将Worker添加到工作线程集合中
                    workers.add(w);
                    int s = workers.size();
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
            // 如果添加成功，启动线程
            if (workerAdded) {
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        // 如果启动失败，回滚操作
        if (!workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}

这段代码主要做了以下几件事：

  1. 检查线程池状态：

    • 首先通过 ctl 这个原子变量获取线程池的状态 rs 和工作线程数量 wc。
    • 然后判断线程池状态，如果线程池已经处于停止状态 SHUTDOWN 及更严重的状态（STOP、TIDYING、TERMINATED），并且不满足特定条件（rs == SHUTDOWN 且 firstTask == null 且任务队列不为空），就直接返回 false，表示不能添加新线程。这是因为在这些状态下，线程池不再接受新任务或者已经在关闭过程中，不适合添加新线程。

  2. 尝试增加工作线程数量：

    • 在一个内层循环中，检查当前工作线程数量是否超过了允许的最大数量（CAPACITY）或者根据 core 参数判断是否超过核心线程数（corePoolSize）或最大线程数（maximumPoolSize）。如果超过了，就返回 false。
    • 接着使用 compareAndIncrementWorkerCount 方法尝试原子性地增加工作线程数量。如果成功（意味着当前线程成功抢到了添加线程的操作权），就跳出外层循环。如果失败，说明有其他线程也在进行类似操作，重新获取线程池状态并继续尝试。

  3. 创建并启动工作线程：

    • 如果成功通过前面的检查和增加线程数量操作，就创建一个新的 Worker 对象，将传入的 firstTask 作为该工作线程的第一个任务。
    • 然后获取 Worker 中的线程 t，对 mainLock 加锁，再次检查线程池状态.

总结

  • 优先尝试将任务放入队列（当线程数已达核心线程数时）。
  • 入队后必须二次检查线程池状态（避免线程池在入队后关闭，导致任务无法处理）。
  • 若入队失败（队列满），则尝试创建非核心线程执行任务。
  • 若所有方式都失败（线程池关闭或线程数达上限），则触发拒绝策略

简单说，就是线程池在「核心线程满 → 队列满 → 非核心线程满」的递进过程中，对任务的处理逻辑，确保任务要么被执行，要么被合理拒绝。

线程池中的钩子方法

线程池中，为了方便后续定位问题，可以在每个任务执行的前后添加环绕逻辑。或者在拒绝策略中添加环绕逻辑，突显拒绝策略被触发了。

import java.util.concurrent.*;

public class ThreadPoolHookDemo extends ThreadPoolExecutor {
  public ThreadPoolHookDemo(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
    super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
  }
  
  @Override
  protected void beforeExecute(Thread t, Runnable r) {
    System.out.println("开始执行任务");
    super.beforeExecute(t, r);
  }

  @Override
  protected void afterExecute(Runnable r, Throwable t) {
    System.out.println("任务执行完成");
    super.afterExecute(r, t);
  }
}

也就是说只要继承了ThreadPoolExecutor类，重写beforeExecute和afterExecute方法，源代码中的runWorker方法中已经嵌入了这两个钩子方法：

    final void runWorker(Worker w) {
  		  // 省略无关代码			
      try {
           beforeExecute(wt, task); // 前置钩子

           } finally {
               afterExecute(task, thrown); // 后置钩子
           }
       } finally {
           task = null;
           w.completedTasks++;
           w.unlock();
       }
   }
   completedAbruptly = false;

而拒绝策略并没有钩子方法，但是仍然可以通过重写相关方法来实现相似的功能：

import java.util.concurrent.ThreadPoolExecutor;

public class AbortPolicyDemo extends ThreadPoolExecutor.AbortPolicy {
  public AbortPolicyDemo() {
    super();
  }

  @Override
  public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    System.out.println("触发拒绝策略");
  }
}

创建了自定义的拒绝策略类后，只需要在构造函数中传入该类的实例即可。

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

  public class ThreadPoolDemo {
    public static void main(String[] args) {
      ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 3, 10,
        TimeUnit.SECONDS, new ArrayBlockingQueue<>(3),
        Executors.defaultThreadFactory(), new AbortPolicyDemo()
        );
      for (int i = 0; i < 20; i++) {
        executor.execute(new Thread(()-> {
          System.out.println(Thread.currentThread().getName() + " 执行任务中");
        }));
      }
    }
  }

执行结果:

触发拒绝策略
pool-1-thread-3 执行任务中
pool-1-thread-1 执行任务中
pool-1-thread-2 执行任务中
pool-1-thread-1 执行任务中
触发拒绝策略
pool-1-thread-1 执行任务中
pool-1-thread-2 执行任务中
pool-1-thread-1 执行任务中
pool-1-thread-2 执行任务中
pool-1-thread-3 执行任务中
pool-1-thread-2 执行任务中
触发拒绝策略
pool-1-thread-1 执行任务中
触发拒绝策略
pool-1-thread-2 执行任务中
pool-1-thread-3 执行任务中
pool-1-thread-2 执行任务中
pool-1-thread-1 执行任务中
pool-1-thread-3 执行任务中

输出结果中看到，一共是输出了20行，因为拒绝策略中我继承的是AbortPolicy，一般业务场景，建议考虑继承CallerRunsPolicy,这样可以保证任务不被丢弃，而且能缓解提交任务的速度。

Studying will never be ending.

▲如有纰漏，烦请指正~~

# java # python # 算法
Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

SeedVR2终极指南：如何用单步扩散模型实现专业级视频修复

还在为老旧视频的模糊画质而烦恼吗？面对抖动严重的监控录像或褪色的家庭影像，你是否曾希望有一种简单高效的修复方案？SeedVR2正是为解决这些痛点而生的一站式视频修复工具。作为字节跳动推出的先进视频修复模型，它能够在单步推理中实现专业级的视频质量提升，彻底改变了传统视频修复的工作流程。## 核心价值：为什么选择SeedVR2进行视频修复？**传统视频修复的三大瓶颈**：1. **计算成本高

avatar AI Agent技术社区

深度解析Amlogic S9xxx电视盒子Armbian系统：从闲置设备到Linux服务器的终极实战指南

你是否拥有一台闲置的Amlogic S9xxx系列电视盒子？想让它摆脱单一的视频播放功能，变身为功能强大的Linux服务器、开发平台或家庭媒体中心？本指南将为你详细解析如何通过Armbian系统实现这一转变，让你的电视盒子重获新生，成为高性能的Linux设备。Armbian系统基于Debian/Ubuntu构建，专为ARM芯片优化，提供了完整的Linux生态支持。## 🎯 核心价值与独特优势

avatar AI Agent技术社区

CANN/runtime Event管理API

本章节描述 CANN Runtime 的 Event 管理接口，用于事件的创建、记录、同步、计时及 IPC 跨进程共享。- [`aclError aclrtCreateEvent(aclrtEvent *event)`](#aclrtCreateEvent)：创建Event，创建出来的Event可用于统计两个Event之间的耗时、多Stream之间的任务同步等场景。- [`aclError

avatar AI Agent技术社区