并发编程 - ThreadPoolExecutor 源码与实战

约 12475 字大约 42 分钟

2026-05-25

在开始以前，先思考下什么是线程创建有哪几种方式，为了避免频繁创建和销毁线程造成不必要的性能，一般在使用线程时，会采用线程池，那什么是线程池呢？

线程池

Java线程池是并发编程的核心组件，而 ThreadPoolExecutor 作为其最核心的实现，通过池化技术和精细化任务调度，解决了频繁创建/销毁线程的资源消耗问题。

本文将从设计思想、工作原理到源码实现，全方位解析 ThreadPoolExecutor 的底层机制。

JDK自带的构建线程池的方式 Executors

JDK 中基于 Executors 提供了很多种线程池

固定大小的线程池 (newFixedThreadPool)

创建一个固定大小的线程池，可控制线程最大并发数，超出的线程会在队列中等待。

public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
            0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<Runnable>());
}

构建时，需要给 ewFixedThreadPool方法提供一个 Threads的属性，而这个属性其实就是当前线程池中线程的个数。当前线程池的本质其实就是使用 ThreadPoolExecutor。

构建好当前线程池后，线程个数已经固定好。如果线程没构建，线程会等待着任务执行时被创建（线程是懒加载，在构建之初，线程并没有构建出来，而是随着人任务的提交才会将线程在线程池中构建出来）。

如果线程都已经构建好了，此时任务会被放到 LinkedBlockingQueue 无界队列中存放，等待线程从LinkedBlockingQueue 中去 take 出任务，然后执行。

public static void main(String[] args) throws Exception {
    ExecutorService threadPool = Executors.newFixedThreadPool(3);
    threadPool.execute(() -> {
        System.out.println("1号任务：" + Thread.currentThread().getName() + System.currentTimeMillis());
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    });
    threadPool.execute(() -> {
        System.out.println("2号任务：" + Thread.currentThread().getName() + System.currentTimeMillis());
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    });
    threadPool.execute(() -> {
        System.out.println("3号任务：" + Thread.currentThread().getName() + System.currentTimeMillis());
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    });
}

单线程的线程池 (newSingleThreadExecutor)

这个线程池看名字就知道是 单例线程池，线程池中只有一个工作线程在处理任务。当你希望所有的任务都在一个线程中顺序执行时使用。这可以保证任务的顺序执行，适用于需要串行处理任务的场景。

// 当前这里就是构建单例线程池的方式
public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
        // 在内部依然是构建了ThreadPoolExecutor，设置的线程个数为1
        // 当任务投递过来后，第一个任务会被工作线程处理，后续的任务会被扔到阻塞队列中
        // 投递到阻塞队列中任务的顺序，就是工作线程处理的顺序
        // 当前这种线程池可以用作顺序处理的一些业务中
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()));
}

示例代码：

public static void main(String[] args) throws Exception {
    ExecutorService threadPool = Executors.newSingleThreadExecutor();

    threadPool.execute(() -> {
        System.out.println(Thread.currentThread().getName() + "," + "111");
    });
    threadPool.execute(() -> {
        System.out.println(Thread.currentThread().getName() + "," + "222");
    });
    threadPool.execute(() -> {
        System.out.println(Thread.currentThread().getName() + "," + "333");
    });
    threadPool.execute(() -> {
        System.out.println(Thread.currentThread().getName() + "," + "444");
    });
}

可缓存的线程池 (newCachedThreadPool)

适用于执行大量短期异步任务的情况。线程池会根据需要创建新线程，并且在长时间闲置后自动终止线程。适用于执行大量独立且耗时短的任务。

public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}

最大的一个特点，**任务只要提交到当前的 ewCachedThreadPool中，就必然有工作线程可以处理**,其本质是SynchronousQueue` 不存储任何元素，直接传递，是一个 生产者-消费者 模型，这也是线程数量设置无界的原因。

当第一次提交任务到线程池时，会直接构建一个工作线程
这个工作线程带执行完后，默认60秒没有任务可以执行后，会结束
如果在等待60秒期间有任务进来，他会再次拿到这个任务去执行
如果后续提升任务时，没有线程是空闲的，那么就构建工作线程去执行。

代码示例：

public static void main(String[] args) throws Exception {
    ExecutorService executorService = Executors.newCachedThreadPool();
    for (int i = 1; i <= 200; i++) {
        final int j = i;
        executorService.execute(() -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread().getName() + ":" + j);
        });
    }
}

定时或周期性任务的线程池 (newScheduledThreadPool)

用于需要定时或定期执行任务的场景。你可以使用 schedule 或 scheduleAtFixedRate 方法来安排任务的执行。

来看一下源码中如何构建的：

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
    return new ScheduledThreadPoolExecutor(corePoolSize);
}

基于这个方法可以看到，构建的是 ScheduledThreadPoolExecutor 线程池

public class ScheduledThreadPoolExecutor extends ThreadPoolExecutor{
	//....
}

所以本质上还是正常线程池，只不过在原来的线程池基础上实现了定时任务的功能。

原理是基于 DelayQueue 实现的延迟执行，周期性执行是任务执行完毕后，再次扔回到阻塞队列。

代码示例：

public static void main(String[] args) throws Exception {
    ScheduledExecutorService pool = Executors.newScheduledThreadPool(10);

    // 正常执行
//        pool.execute(() -> {
//            System.out.println(Thread.currentThread().getName() + "：1");
//        });

    // 延迟执行，执行当前任务延迟5s后再执行
//        pool.schedule(() -> {
//            System.out.println(Thread.currentThread().getName() + "：2");
//        },5,TimeUnit.SECONDS);

    // 周期执行，当前任务第一次延迟5s执行，然后没3s执行一次
    // 这个方法在计算下次执行时间时，是从任务刚刚开始时就计算。
//        pool.scheduleAtFixedRate(() -> {
//            try {
//                Thread.sleep(3000);
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//            System.out.println(System.currentTimeMillis() + "：3");
//        },2,1,TimeUnit.SECONDS);

    // 周期执行，当前任务第一次延迟5s执行，然后没3s执行一次
    // 这个方法在计算下次执行时间时，会等待任务结束后，再计算时间
    pool.scheduleWithFixedDelay(() -> {
        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(System.currentTimeMillis() + "：3");
    },2,1,TimeUnit.SECONDS);
}

至于 Executors 提供的 ewSingleThreadScheduledExecutor` 单例的定时任务线程池就不说了，一个线程的线程池可以延迟或者以一定的周期执行一个任务。

参数	设置准则	典型值示例
corePoolSize	CPU密集型：CPU核心数； IO 密集型：CPU核心数 * (1 + 平均等待时间 W/计算时间 C)	8核CPU：IO密集型可设32
maximumPoolSize	核心线程数 * 2 ~ 3倍	core=8 → max=24
keepAliveTime	非核心线程存活时间，建议30~60秒	30, TimeUnit.SECONDS

队列类型	特点	适用场景
SynchronousQueue	零容量，直接传递任务	高吞吐量短任务
LinkedBlockingQueue	无界队列（默认Integer.MAX_VALUE）	需积压任务的批处理系统
ArrayBlockingQueue	有界队列，FIFO	流量可控的实时系统
PriorityBlockingQueue	优先级队列	任务有优先级的调度系统

ScheduledThreadPoolExecutor

从名字上就可以看出，当前线程池是用于执行定时任务的线程池。

Java比较早的定时任务工具是 Timer 类。但是 Timer 问题很多，串行的，不靠谱，会影响到其他的任务执行。

其实除了 Timer 以及 ScheduleThreadPoolExecutor 之外，正常在企业中一般会采用 Quartz 或者是 SpringBoot 提供的 Schedule 的方式去实现定时任务的功能。

ScheduledThreadPoolExecutor 支持延迟执行以及周期性执行的功能。

ScheduledThreadPoolExecutor 应用

定时任务线程池的有参构造

public ScheduledThreadPoolExecutor(int corePoolSize,
                                   ThreadFactory threadFactory,
                                   RejectedExecutionHandler handler) {
    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
          new DelayedWorkQueue(), threadFactory, handler);
}

发现 ScheduledThreadPoolExecutor 在构建时，直接调用了父类的构造方法

ScheduledThreadPoolExecutor 的父类就是 ThreadPoolExecutor

首先 ScheduleThreadPoolExecutor 最多允许设置3个参数：

核心线程数
线程工厂
拒绝策略

首先没有设置阻塞队列，以及最大线程数和空闲时间以及单位

阻塞队列设置的是 DelayedWorkQueue，其实本质就是 DelayQueue，一个延迟队列。DelayQueue 是一个无界队列。所以最大线程数以及非核心线程的空闲时间是不需要设置的。

代码落地使用

public static void main(String[] args) {
        //1. 构建定时任务线程池
        ScheduledThreadPoolExecutor pool = new ScheduledThreadPoolExecutor(
                5,
                new ThreadFactory() {
                    @Override
                    public Thread newThread(Runnable r) {
                        Thread t = new Thread(r);
                        return t;
                    }
                },
                new ThreadPoolExecutor.AbortPolicy()
        );
  
        //2. 应用ScheduledThreadPoolExecutor
        // 跟直接执行线程池的execute没啥区别
        pool.execute(() -> {
            System.out.println("execute");
        });
  
        // 指定延迟时间执行
        System.out.println(System.currentTimeMillis());
        pool.schedule(() -> {
            System.out.println("schedule");
            System.out.println(System.currentTimeMillis());
        },2, TimeUnit.SECONDS);
  
        // 指定第一次的延迟时间，并且确认后期的周期执行时间，周期时间是在任务开始时就计算
        // 周期性执行就是将执行完毕的任务再次设置好延迟时间，并且重新扔到阻塞队列
        // 计算的周期执行，也是在原有的时间上做累加，不关注任务的执行时长。
        System.out.println(System.currentTimeMillis());
        pool.scheduleAtFixedRate(() -> {
            System.out.println("scheduleAtFixedRate");
            System.out.println(System.currentTimeMillis());
        },2,3,TimeUnit.SECONDS);
  
  
        // 指定第一次的延迟时间，并且确认后期的周期执行时间，周期时间是在任务结束后再计算下次的延迟时间
        System.out.println(System.currentTimeMillis());
        pool.scheduleWithFixedDelay(() -> {
            System.out.println("scheduleWithFixedDelay");
            System.out.println(System.currentTimeMillis());
            try {
                Thread.sleep(4000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },2,3,TimeUnit.SECONDS);
    }

ScheduledThreadPoolExecutor 源码剖析

核心属性

后面的方法业务流程会涉及到这些属性。

// 这里是针对任务取消时的一些业务判断会用到的标记
private volatile boolean continueExistingPeriodicTasksAfterShutdown;
private volatile boolean executeExistingDelayedTasksAfterShutdown = true;
private volatile boolean removeOnCancel = false;

// 计数器，如果两个任务的执行时间节点一模一样，根据这个序列来判断谁先执行
private static final AtomicLong sequencer = new AtomicLong();

// 这个方法是获取当前系统时间的毫秒值
final long now() {
    return System.nanoTime();
}

// 内部类。核心类之一。
private class ScheduledFutureTask<V>
        extends FutureTask<V> implements RunnableScheduledFuture<V> {

    // 全局唯一的序列，如果两个任务时间一直，基于当前属性判断
    private final long sequenceNumber;

    // 任务执行的时间，单位纳秒
    private long time;

    /**
     *  period == 0：执行一次的延迟任务
     *  period > 0：代表是At
     *  period < 0：代表是With
     */
    private final long period;

    // 周期性执行时，需要将任务重新扔回阻塞队列，基础当前属性拿到任务，方便扔回阻塞队列
    RunnableScheduledFuture<V> outerTask = this;


    /**
     * 构建schedule方法的任务
     */
    ScheduledFutureTask(Runnable r, V result, long ns) {
        super(r, result);
        this.time = ns;
        this.period = 0;
        this.sequenceNumber = sequencer.getAndIncrement();
    }

    /**
     * 构建At和With任务的有参构造
     */  
    ScheduledFutureTask(Runnable r, V result, long ns, long period) {
        super(r, result);
        this.time = ns;
        this.period = period;
        this.sequenceNumber = sequencer.getAndIncrement();
    }

}   

// 核心组件：延时阻塞队列（DelayedWorkQueue）
private class DelayedWorkQueue extends AbstractQueue<Runnable>
    implements BlockingQueue<Runnable> {
    // 基于最小堆（优先队列）管理待执行任务
    private RunnableScheduledFuture<?>[] queue = new RunnableScheduledFuture<?>[16];
    // 堆排序保证队首任务过期时间最近
}

时间驱动调度：任务封装为 ScheduledFutureTask，记录下次执行时间 time（纳秒精度）
任务排序：队列按 time 排序，队首任务最先到期（peek().getDelay() <= 0）

schedule 方法

execute 方法也是调用的 schedule 方法，只不过传入的延迟时间是0纳秒

schedule 方法就是将任务和延迟时间封装到一起，并且将任务扔到阻塞队列中，再去创建工作线程去 take 阻塞队列。

// 延迟任务执行的方法。
// command：任务
// delay：延迟时间
// unit：延迟时间的单位
public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
    // 健壮性校验。
    if (command == null || unit == null)
        throw new NullPointerException();

    // 将任务和延迟时间封装到一起，最终组成ScheduledFutureTask
    // 要分成三个方法去看
    // triggerTime：计算延迟时间。最终返回的是当前系统时间 + 延迟时间 
    // triggerTime就是将延迟时间转换为纳秒，并且+当前系统时间，再做一些健壮性校验

    // ScheduledFutureTask有参构造：将任务以及延迟时间封装到一起，并且设置任务执行的方式

    // decorateTask：当前方式是让用户基于自身情况可以动态修改任务的一个扩展口
    RunnableScheduledFuture<?> t = decorateTask(command, 
                                   new ScheduledFutureTask<Void>(command, null,
                                   triggerTime(delay, unit)));
    // 任务封装好，执行delayedExecute方法，去执行任务
    delayedExecute(t);

    // 返回FutureTask
    return t;
}

// triggerTime做的事情
// 外部方法，对延迟时间做校验，如果小于0，就直接设置为0
// 并且转换为纳秒单位
private long triggerTime(long delay, TimeUnit unit) {
    return triggerTime(unit.toNanos((delay < 0) ? 0 : delay));
}
// 将延迟时间+当前系统时间
// 后面的校验是为了避免延迟时间超过Long的取值范围
long triggerTime(long delay) {
    return now() + ((delay < (Long.MAX_VALUE >> 1)) ? delay : overflowFree(delay));
}

// ScheduledFutureTask有参构造
ScheduledFutureTask(Runnable r, V result, long ns) {
    super(r, result);
    // time就是任务要执行的时间
    this.time = ns;
    // period,为0，代表任务是延迟执行，不是周期执行
    this.period = 0;
    // 基于AtmoicLong生成的序列
    this.sequenceNumber = sequencer.getAndIncrement();
}


// delayedExecute 执行延迟任务的操作
private void delayedExecute(RunnableScheduledFuture<?> task) {
    // 查看当前线程池是否还是RUNNING状态，如果不是RUNNING，进到if
    if (isShutdown())
        // 不是RUNNING。
        // 执行拒绝策略。
        reject(task);
    else {
        // 线程池状态是RUNNING
        // 直接让任务扔到延迟的阻塞队列中
        super.getQueue().add(task);
        // DCL的操作，再次查看线程池状态
        // 如果线程池在添加任务到阻塞队列后，状态不是RUNNING
        if (isShutdown() &&
            // task.isPeriodic()：现在反回的是false，因为任务是延迟执行，不是周期执行
            // 默认情况，延迟队列中的延迟任务，可以执行
            !canRunInCurrentRunState(task.isPeriodic()) &&
            // 从阻塞队列中移除任务。
            remove(task))
            task.cancel(false);
        else
            // 线程池状态正常，任务可以执行
            ensurePrestart();
    }
}

// 线程池状态不为RUNNING，查看任务是否可以执行
// 延迟执行：periodic==false
// 周期执行：periodic==true
// continueExistingPeriodicTasksAfterShutdown：周期执行任务，默认为false
// executeExistingDelayedTasksAfterShutdown：延迟执行任务，默认为true
boolean canRunInCurrentRunState(boolean periodic) {
    return isRunningOrShutdown(periodic ?
                               continueExistingPeriodicTasksAfterShutdown :
                               executeExistingDelayedTasksAfterShutdown);
}
// 当前情况，shutdownOK为true
final boolean isRunningOrShutdown(boolean shutdownOK) {
    int rs = runStateOf(ctl.get());
    // 如果状态是RUNNING，正常可以执行，返回true
    // 如果状态是SHUTDOWN，根据shutdownOK来决定
    return rs == RUNNING || (rs == SHUTDOWN && shutdownOK);
}


// 任务可以正常执行后，做的操作
void ensurePrestart() {
    // 拿到工作线程个数
    int wc = workerCountOf(ctl.get());
    // 如果工作线程个数小于核心线程数
    if (wc < corePoolSize)
        // 添加核心线程去处理阻塞队列中的任务
        addWorker(null, true);
    else if (wc == 0)
        // 如果工作线程数为0，核心线程数也为0，这是添加一个非核心线程去处理阻塞队列任务
        addWorker(null, false);
}

At 和 With 方法 & 任务的 run 方法

scheduleAtFixedRate 和 scheduleWithFixedDelay 的差异主要体现在任务间隔计算上。

前者像地铁时刻表，固定发车间隔；后者更像每次任务完成后才开始倒计时下次执行。

这个区别在任务执行时间超过间隔时会产生截然不同的行为：At会堆积任务，With会顺延。它们的适用场景，比如数据同步用At，资源清理用With。

这两个方法在源码层面上的第一个区别，就是在计算周期时间时，需要将这个值传递给period，基于正负数在区别At和With

所以查看一个方法就ok，查看At方法

// At方法，
// command：任务
// initialDelay：第一次执行的延迟时间
// period：任务的周期执行时间
// unit：上面两个时间的单位
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
                                              long initialDelay,
                                              long period,
                                              TimeUnit unit) {
    // 健壮性校验
    if (command == null || unit == null)
        throw new NullPointerException();
    // 周期时间不能小于等于0.
    if (period <= 0)
        throw new IllegalArgumentException();
    // 将任务以及第一次的延迟时间，和后续的周期时间封装好。
    ScheduledFutureTask<Void> sft =
        new ScheduledFutureTask<Void>(command,
                                      null,
                                      triggerTime(initialDelay, unit),
                                      unit.toNanos(period));
    // 扩展口，可以对任务做修改。
    RunnableScheduledFuture<Void> t = decorateTask(command, sft);

    // 周期性任务，需要在任务执行完毕后，重新扔会到阻塞队列，为了方便拿任务，将任务设置到outerTask成员变量中
    sft.outerTask = t;
    // 和schedule方法一样的方式
    // 如果任务刚刚扔到阻塞队列，线程池状态变为SHUTDOWN，默认情况，当前任务不执行
    delayedExecute(t);
    return t;
}

// 延迟任务以及周期任务在执行时，都会调用当前任务的run方法。
public void run() {
    // periodic == false：一次性延迟任务
    // periodic == true：周期任务
    boolean periodic = isPeriodic();
    // 任务执行前，会再次判断状态，能否执行任务
    if (!canRunInCurrentRunState(periodic))
        cancel(false);
    // 判断是周期执行还是一次性任务
    else if (!periodic)
        // 一次性任务，让工作线程直接执行command的逻辑
        ScheduledFutureTask.super.run();
    // 到这个else if，说明任务是周期执行
    else if (ScheduledFutureTask.super.runAndReset()) {
        // 设置下次任务执行的时间
        setNextRunTime();
        // 将任务重新扔回线程池做处理
        reExecutePeriodic(outerTask);
    }
}
// 设置下次任务执行的时间
private void setNextRunTime() {
    // 拿到period值，正数：At，负数：With
    long p = period;
    if (p > 0)
        // 拿着之前的执行时间，直接追加上周期时间
        time += p;
    else
        // 如果走到else，代表任务是With方式，这种方式要重新计算延迟时间
        // 拿到当前系统时间，追加上延迟时间，
        time = triggerTime(-p);
}
 // 将任务重新扔回线程池做处理
void reExecutePeriodic(RunnableScheduledFuture<?> task) {
    // 如果状态ok，可以执行
    if (canRunInCurrentRunState(true)) {
        // 将任务扔到延迟队列
        super.getQueue().add(task);
        // DCL，判断线程池状态
        if (!canRunInCurrentRunState(true) && remove(task))
            task.cancel(false);
        else
            // 添加工作线程
            ensurePrestart();
    }
}

总结

ThreadPoolExecutor 的核心价值在于通过三层缓冲机制实现资源的最优调度：

核心线程池：常驻线程处理稳态流量
任务队列：缓冲突发流量
非核心线程：应对队列满载的极端场景

配合状态机和 CAS无锁操作，实现了高并发环境下的线程安全调度。其源码中展现的位运算技巧（如ctl设计，存储工作线程数量+线程池状态）和 Worker 线程复用机制，是Java并发编程的经典范例。

性能数据参考： 在8核服务器上，合理配置的 ThreadPoolExecutor 可支撑20,000+ QPS的任务调度，相比单线程执行效率提升8-10倍，同时保持95%的CPU利用率。

ScheduledThreadPoolExecutor 继承了 ThreadPoolExecutor，且定时执行是基于延迟队列 DelayedWorkQueue 实现的，底层任务是基于 ScheduledFutureTask 优先队列，内部记录下次执行时间 time，队列按 time 排序，队首任务最先到期的。At/With 方法主要是通过 period 参数控制，是执行一次，还是按固定时间或固定延迟时间执行, At 会堆积任务，With 会顺延。融合了 时间轮算法思想（通过优先队列模拟）和 线程池资源管理，是 Java 并发库中高精度调度的经典设计

掌握 ThreadPoolExecutor 的原理与实现，是构建高并发、高可靠Java应用的基石。建议结合 ForkJoinPool（分治场景）和 ScheduledThreadPoolExecutor（定时任务）形成完整的线程池技术体系。

版权所有

版权归属：今朝· Liushigong · Relax Box

许可证：署名-非商业性 4.0 国际 (CC-BY-NC-4.0)

并发编程 - ThreadPoolExecutor 源码与实战

线程池

JDK自带的构建线程池的方式 Executors

固定大小的线程池 (newFixedThreadPool)

单线程的线程池 (newSingleThreadExecutor)

可缓存的线程池 (newCachedThreadPool)

定时或周期性任务的线程池 (newScheduledThreadPool)

ThreadPoolExecutor

为什么要自定义线程池

ThreadPoolExecutor 应用

ThreadPoolExecutor 源码剖析

核心属性

有参构造

execute 方法

addWorker 方法

Worker 工作线程

ThreadPoolExecutor 的 runWorker 方法

getTask 方法

shutdownNow 关闭方法

线程池的核心参数设计规则

线程池参数示例

队列选型指南

线程池处理任务的核心流程

ScheduledThreadPoolExecutor

ScheduledThreadPoolExecutor 应用

ScheduledThreadPoolExecutor 源码剖析

核心属性

schedule 方法

At 和 With 方法 & 任务的 run 方法

总结

版权所有