Java Executors 框架源码深度分析(上)

我真的是大笨蛋

1185人浏览 · 2025-09-22 20:46:11

我真的是大笨蛋 · 2025-09-22 20:46:11 发布

概述

java.util.concurrent.Executors 是 Java 并发包中最重要的工厂类之一，它提供了一系列创建线程池和执行器的静态工厂方法。是并发框架的重要组成部分，Executors 类简化了线程池的创建和使用，是 Java 并发编程的基石。

类结构设计分析

线程池设计模式

Executors 类采用了多种设计模式：

工厂方法模式

每个 newXXX 方法都是工厂方法的典型实现，隐藏了具体实现类的创建细节。
在 Executors 中，工厂方法模式被大量用于创建不同类型的线程池，隐藏了复杂的构造逻辑。

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


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

优势

用户只需提供线程数量，无需关心 ThreadPoolExecutor 的复杂参数配置
返回 ExecutorService 接口，隐藏具体实现类

装饰器模式

通过 DelegatedExecutorService 等包装类，限制对底层实现特定方法的访问。在 Executors 中，装饰器模式主要用于限制对线程池配置的修改，增强安全性。

private static class DelegatedExecutorService implements ExecutorService {
    private final ExecutorService e;
    
    DelegatedExecutorService(ExecutorService executor) { 
        e = executor; 
    }
    
    // 委托所有 ExecutorService 接口方法
    public void execute(Runnable command) {
        try {
            e.execute(command);
        } finally { 
            reachabilityFence(this); 
        }
    }
    
    public void shutdown() {
        try {
            e.shutdown();
        } finally { 
            reachabilityFence(this); 
        }
    }
    
    // 其他方法委托实现...
}

优势

防止不安全的配置修改, 限制对底层实现的直接访问
动态添加新功能（如自动关闭）, 保持接口兼容性

线程池体系

Executors 的线程池主要分为以下几类：

固定大小线程池（FixedThreadPool）
缓存线程池（CachedThreadPool）
单线程执行器（SingleThreadExecutor）
定时任务线程池（ScheduledThreadPool）
工作窃取线程池（WorkStealingPool）
每任务线程执行器（ThreadPerTaskExecutor）

核心线程池实现分析

3.1 固定大小线程池（FixedThreadPool）

先来查看一下源码中的介绍

    /**
     * Creates a thread pool that reuses a fixed number of threads
     * operating off a shared unbounded queue.  At any point, at most
     * {@code nThreads} threads will be active processing tasks.
     * If additional tasks are submitted when all threads are active,
     * they will wait in the queue until a thread is available.
     * If any thread terminates due to a failure during execution
     * prior to shutdown, a new one will take its place if needed to
     * execute subsequent tasks.  The threads in the pool will exist
     * until it is explicitly {@link ExecutorService#shutdown shutdown}.
     *
     * @param nThreads the number of threads in the pool
     * @return the newly created thread pool
     * @throws IllegalArgumentException if {@code nThreads <= 0}
     */

翻译

创建一个线程池，该线程池重用固定数量的线程来处理共享的无界队列。在任何时候，最多只有 nThreads 个线程处于活动状态来处理任务。如果所有线程都处于活动状态时提交额外的任务，这些任务将在队列中等待，直到有线程可用。如果在关闭前执行期间任何线程因故障而终止，如果需要执行后续任务，将创建新线程来替代它。线程池中的线程将持续存在，直到显式调用 ExecutorService#shutdown 方法关闭线程池。

底层实现

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

核心参数：

corePoolSize 和 maximumPoolSize 相等，保持固定线程数
keepAliveTime 为 0，由于核心线程数等于最大线程数，此参数无效
使用无界队列 LinkedBlockingQueue，可能导致内存溢出

工作队列：
在这里插入图片描述

潜在问题

无界队列可能导致 OOM：当任务提交速度远大于处理速度时，队列无限增长
不适合执行耗时较长的任务：容易造成任务堆积

缓存线程池（CachedThreadPool）

    /**
     * Creates a thread pool that creates new threads as needed, but
     * will reuse previously constructed threads when they are
     * available.  These pools will typically improve the performance
     * of programs that execute many short-lived asynchronous tasks.
     * Calls to {@code execute} will reuse previously constructed
     * threads if available. If no existing thread is available, a new
     * thread will be created and added to the pool. Threads that have
     * not been used for sixty seconds are terminated and removed from
     * the cache. Thus, a pool that remains idle for long enough will
     * not consume any resources. Note that pools with similar
     * properties but different details (for example, timeout parameters)
     * may be created using {@link ThreadPoolExecutor} constructors.
     *
     * @return the newly created thread pool
     */

翻译

创建一个根据需要创建新线程的线程池，但当有可用线程时，会重用先前构造的线程。这类线程池通常可以提高执行许多短期异步任务的程序的性能。对 execute 的调用在可能的情况下会重用先前构造的线程。如果不存在可用线程，则会创建一个新线程并添加到线程池中。未被使用超过六十秒的线程将被终止并从缓存中移除。因此，长时间处于空闲状态的线程池不会消耗任何资源。注意，可以使用 ThreadPoolExecutor 构造函数创建具有相似属性但细节不同（例如超时参数）的线程池。

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

参数分析

核心线程数为 0，全部为临时线程
最大线程数为 Integer.MAX_VALUE，理论上可创建无限线程
线程空闲 60 秒后回收
使用 SynchronousQueue，不存储任务，直接传递

SynchronousQueue 工作机制
源码
在这里插入图片描述

SynchronousQueue默认是非公平的
Transferer<E>是支持 Lifo（栈）模式的 LinkedTransferQueue 扩展
每个插入的任务操作必须等待对应的移除操作

适用场景

大量短期异步任务
任务处理时间较短
对线程创建销毁开销不敏感的场景

单线程执行器（SingleThreadExecutor）

    /**
     * Creates a single-threaded executor that can schedule commands
     * to run after a given delay, or to execute periodically.
     * (Note however that if this single
     * thread terminates due to a failure during execution prior to
     * shutdown, a new one will take its place if needed to execute
     * subsequent tasks.)  Tasks are guaranteed to execute
     * sequentially, and no more than one task will be active at any
     * given time. Unlike the otherwise equivalent
     * {@code newScheduledThreadPool(1)} the returned executor is
     * guaranteed not to be reconfigurable to use additional threads.
     *
     * @return the newly created scheduled executor
     */

翻译

创建一个单线程执行器，可以调度命令在给定延迟后运行，或定期执行。（请注意，如果这个单线程在执行过程中因故障而终止，在关闭之前，如果需要执行后续任务，将会有一个新的线程取而代之。）任务保证顺序执行，在任何给定时间最多只有一个任务处于活动状态。与等效的 newScheduledThreadPool(1) 不同，返回的执行器保证不会重新配置为使用额外的线程。

 public static ScheduledExecutorService newSingleThreadScheduledExecutor() {
        return new DelegatedScheduledExecutorService
            (new ScheduledThreadPoolExecutor(1));
    }

特殊设计

使用 AutoShutdownDelegatedExecutorService 包装，防止被重新配置
保证任务顺序执行
线程异常终止后自动创建新线程

与 FixedThreadPool(1) 的区别：

// 普通 FixedThreadPool(1) 可以强制转型并修改配置
ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(1);
executor.setCorePoolSize(2);  // 可以修改

// SingleThreadExecutor 不可重新配置
ExecutorService singleExecutor = Executors.newSingleThreadExecutor();
// 无法获取到底层的 ThreadPoolExecutor 进行修改

定时任务线程池（ScheduledThreadPool）

    /**
     * Creates a single-threaded executor that can schedule commands
     * to run after a given delay, or to execute periodically.  (Note
     * however that if this single thread terminates due to a failure
     * during execution prior to shutdown, a new one will take its
     * place if needed to execute subsequent tasks.)  Tasks are
     * guaranteed to execute sequentially, and no more than one task
     * will be active at any given time. Unlike the otherwise
     * equivalent {@code newScheduledThreadPool(1, threadFactory)}
     * the returned executor is guaranteed not to be reconfigurable to
     * use additional threads.
     *
     * @param threadFactory the factory to use when creating new threads
     * @return the newly created scheduled executor
     * @throws NullPointerException if threadFactory is null
     */

翻译

创建一个单线程执行器，可以调度命令在给定延迟后运行，或定期执行。（注意，如果这个单线程在执行过程中因故障而终止，在关闭之前，如果需要执行后续任务，将会有新的线程取而代之。）任务保证顺序执行，在任何给定时间最多只有一个任务处于活动状态。与等效的 newScheduledThreadPool(1, threadFactory) 不同，返回的执行器保证不会重新配置为使用额外的线程。

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

核心实现
ScheduledThreadPoolExecutor 继承自 ThreadPoolExecutor，重写了任务调度逻辑。

任务队列特殊性

// 使用特殊的 DelayedWorkQueue
static class DelayedWorkQueue extends AbstractQueue<Runnable>
    implements BlockingQueue<Runnable> {
    // 基于堆结构的延迟队列
    private RunnableScheduledFuture<?>[] queue = 
        new RunnableScheduledFuture<?>[INITIAL_CAPACITY];
}

线程工厂（ThreadFactory）机制

默认线程工厂

private static class DefaultThreadFactory implements ThreadFactory {
    private static final AtomicInteger poolNumber = new AtomicInteger(1);
    private final ThreadGroup group;
    private final AtomicInteger threadNumber = new AtomicInteger(1);
    private final String namePrefix;

    DefaultThreadFactory() {
        SecurityManager s = System.getSecurityManager();
        group = (s != null) ? s.getThreadGroup() :
                              Thread.currentThread().getThreadGroup();
        namePrefix = "pool-" + poolNumber.getAndIncrement() + "-thread-";
    }

    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);
        return t;
    }
}

设计要点

线程命名：pool-N-thread-M 格式，便于监控和调试
线程组管理：统一线程组，便于安全管理
标准化配置：非守护线程，标准优先级

特权线程工厂（已废弃）

@Deprecated(since="17", forRemoval=true)
public static ThreadFactory privilegedThreadFactory() {
    return new PrivilegedThreadFactory();
}

安全机制实现

private static class PrivilegedThreadFactory extends DefaultThreadFactory {
    final AccessControlContext acc;
    final ClassLoader ccl;

    PrivilegedThreadFactory() {
        super();
        // 安全检查
        SecurityManager sm = System.getSecurityManager();
        if (sm != null) {
            sm.checkPermission(SecurityConstants.GET_CLASSLOADER_PERMISSION);
            sm.checkPermission(new RuntimePermission("setContextClassLoader"));
        }
        this.acc = AccessController.getContext();
        this.ccl = Thread.currentThread().getContextClassLoader();
    }

    public Thread newThread(final Runnable r) {
        return super.newThread(new Runnable() {
            @SuppressWarnings("removal")
            public void run() {
                AccessController.doPrivileged(new PrivilegedAction<>() {
                    public Void run() {
                        Thread.currentThread().setContextClassLoader(ccl);
                        r.run();
                        return null;
                    }
                }, acc);
            }
        });
    }
}