深入解析 Java 线程池的开源扩展方案

线程池的执行入口

public void execute(Runnable command, long timeout, TimeUnit unit) {
    // 提交的任务数 +1
    this.submittedCount.incrementAndGet();
    try {
        // 调用 JDK 线程池的 execute 方法。重点在 TaskQueue 的 offer 方法
        super.execute(command);
    } catch (RejectedExecutionException rx) {
        if (!(super.getQueue() instanceof TaskQueue)) {
            this.submittedCount.decrementAndGet();
            throw rx;
        }
        // 任务队列已积满任务
        TaskQueue queue = (TaskQueue) super.getQueue();
        try {
            // 再次尝试将任务放入队列
            if (!queue.force(command, timeout, unit)) {
                this.submittedCount.decrementAndGet();
                throw new RejectedExecutionException(sm.getString("threadPoolExecutor.queueFull"));
            }
        } catch (InterruptedException x) {
            // 提交的任务数 -1
            this.submittedCount.decrementAndGet();
            throw new RejectedExecutionException(x);
        }
    }
}

核心队列：TaskQueue（Tomcat 自定义的阻塞队列）

这是 Tomcat 线程池和 JDK 原生最核心的差异点，源码核心逻辑：

public class TaskQueue extends LinkedBlockingQueue<Runnable> {
    private transient ThreadPoolExecutor executor;

    public boolean force(Runnable o, long timeout, TimeUnit unit) throws InterruptedException {
        if (this.parent != null && !this.parent.isShutdown()) {
            // 此处调用父类 LinkedBlockingQueue 的 offer 方法
            return super.offer(o, timeout, unit);
        } else {
            throw new RejectedExecutionException(sm.getString("taskQueue.notRunning"));
        }
    }

    // 重写 offer 方法：Web 场景下，队列满时直接返回 false，触发拒绝策略
    public boolean offer(Runnable o) {
        if (this.parent == null) {
            return super.offer(o);
        } else if (this.parent.getPoolSize() == this.parent.getMaximumPoolSize()) {
            // 如果创建的线程数等于最大线程数，放入任务队列（此时只能排队等待处理）
            return super.offer(o);
        } else if (this.parent.getSubmittedCount() <= this.parent.getPoolSize()) {
            // 提交的任务数小于核心线程数，放入队列（此时核心线程足以处理提交的任务，不需要创建临时线程）
            return super.offer(o);
        } else {
            // 创建的线程数小于最大线程数，返回 false，此时提交的任务不会排队，而是创建临时线程；否则放入队列
            return this.parent.getPoolSize() < this.parent.getMaximumPoolSize() ? false : super.offer(o);
        }
    }
}

源码解读

JDK 原生线程池的逻辑是'核心线程→队列→临时线程→拒绝'，而 Tomcat 通过重写 offer 方法，改成了'核心线程→临时线程（直到 maxThreads）→队列→拒绝'——这意味着 Tomcat 会优先创建线程处理请求，而非让请求排队，完美适配 Web 请求'短、快'的特性，减少请求排队等待时间。

核心扩展：预启动核心线程

Tomcat 线程池支持配置 prestartminSpareThreads="true"，源码层面会在初始化时调用。预启动核心线程能避免'首次请求创建线程的开销'——生产中建议开启，尤其是秒杀、活动等突发流量场景。

public void prestartAllCoreThreads() {
    int n = prestartCoreThread();
    while (n < getCorePoolSize()) { 
        n += prestartCoreThread();
    }
}

二、Dubbo（3.3.0）

与 Tomcat 中线程池的实现类似。

org.apache.dubbo.common.threadpool.support.eager.EagerThreadPool#getExecutor

public Executor getExecutor(URL url) {
    String name = url.getParameter(THREAD_NAME_KEY, DEFAULT_THREAD_NAME);
    int cores = url.getParameter(CORE_THREADS_KEY, DEFAULT_CORE_THREADS);
    int threads = url.getParameter(THREADS_KEY, Integer.MAX_VALUE);
    int queues = url.getParameter(QUEUES_KEY, DEFAULT_QUEUES);
    int alive = url.getParameter(ALIVE_KEY, DEFAULT_ALIVE);
    // 初始化队列和线程池
    TaskQueue<Runnable> taskQueue = new TaskQueue<>(queues <= 0 ? 1 : queues);
    EagerThreadPoolExecutor executor = new EagerThreadPoolExecutor(cores, threads, alive, TimeUnit.MILLISECONDS, taskQueue, new NamedInternalThreadFactory(name, true), new AbortPolicyWithReport(name, url));
    taskQueue.setExecutor(executor);
    return executor;
}

org.apache.dubbo.common.threadpool.support.eager.EagerThreadPoolExecutor#execute

public void execute(Runnable command) {
    if (command == null) {
        throw new NullPointerException();
    }
    // 提交的任务数 +1
    submittedTaskCount.incrementAndGet();
    try {
        super.execute(command);
    } catch (RejectedExecutionException rx) {
        // retry to offer the task into queue
        final TaskQueue queue = (TaskQueue) super.getQueue();
        try {
            // 拒绝之后再次尝试放入队列
            if (!queue.retryOffer(command, 0, TimeUnit.MILLISECONDS)) {
                submittedTaskCount.decrementAndGet();
                // 任务数 -1
                throw new RejectedExecutionException("Queue capacity is full.", rx);
            }
        } catch (InterruptedException x) {
            submittedTaskCount.decrementAndGet();
            throw new RejectedExecutionException(x);
        }
    } catch (Throwable t) {
        // decrease any way
        submittedTaskCount.decrementAndGet();
        throw t;
    }
}

org.apache.dubbo.common.threadpool.support.eager.TaskQueue#offer

public boolean offer(Runnable runnable) {
    if (executor == null) {
        throw new RejectedExecutionException("The task queue does not have executor!");
    }
    // 当前启动的线程数
    int currentPoolThreadSize = executor.getPoolSize();
    // 提交的任务数小于核心线程数，说明核心线程足以处理，放入队列，让核心线程处理
    if (executor.getSubmittedTaskCount() < currentPoolThreadSize) {
        return super.offer(runnable);
    }
    // 已创建的线程数小于最大线程数，放回 false，不会将任务放入队列，而是创建临时线程
    if (currentPoolThreadSize < executor.getMaximumPoolSize()) {
        return false;
    }
    // 直接放入队列，排队
    return super.offer(runnable);
}

默认拒绝策略

org.apache.dubbo.common.threadpool.support.AbortPolicyWithReport#rejectedExecution

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    String msg = String.format("Thread pool is EXHAUSTED! Thread Name: %s, Pool Size: %d (active: %d, core: %d, max: %d, largest: %d), Task: %d (completed: %d), Executor status:(isShutdown:%s, isTerminated:%s, isTerminating:%s), in %s://%s:%d!",
            this.threadName, e.getPoolSize(), e.getActiveCount(), e.getCorePoolSize(), e.getMaximumPoolSize(), e.getLargestPoolSize(), e.getTaskCount(), e.getCompletedTaskCount(), e.isShutdown(), e.isTerminated(), e.isTerminating(), this.url.getProtocol(), this.url.getIp(), this.url.getPort());
    logger.warn("0-1", "too much client requesting provider", "", msg);
    if (Boolean.parseBoolean(this.url.getParameter("dump.enable", Boolean.TRUE.toString()))) {
        this.dumpJStack();
    }
    // 事件监听机制，可扩展 ThreadPoolExhaustedListener，监听 ThreadPoolExhaustedEvent 事件
    this.dispatchThreadPoolExhaustedEvent(msg);
    throw new RejectedExecutionException(msg);
}

三、Motan（1.2.0）

com.weibo.api.motan.core.StandardThreadExecutor

继承 JDK 的线程池，设计思路借鉴于 Tomcat。

队列 ExecutorQueue，继承 LinkedTransferQueue，能保证更高性能，相比 LinkedBlockingQueue 有明显提升。

public StandardThreadExecutor(int coreThreads, int maxThreads, long keepAliveTime, TimeUnit unit, int queueCapacity, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
    // 初始化
    super(coreThreads, maxThreads, keepAliveTime, unit, new ExecutorQueue(), // 任务队列，继承自 LinkedTransferQueue
            threadFactory, handler);
    ((ExecutorQueue) getQueue()).setStandardThreadExecutor(this);
    // 统计提交的任务数
    submittedTasksCount = new AtomicInteger(0);
    // 可以提交的最大任务数：队列长度数 + 最大线程数
    maxSubmittedTaskCount = queueCapacity + maxThreads;
}

public void execute(Runnable command) {
    int count = submittedTasksCount.incrementAndGet();
    // 超过可以提交的任务数，进行 reject
    if (count > maxSubmittedTaskCount) {
        submittedTasksCount.decrementAndGet();
        getRejectedExecutionHandler().rejectedExecution(command, this);
    }
    try {
        super.execute(command);
    } catch (RejectedExecutionException rx) {
        // 再次尝试放入队列
        if (!((ExecutorQueue) getQueue()).force(command)) {
            submittedTasksCount.decrementAndGet();
            getRejectedExecutionHandler().rejectedExecution(command, this);
        }
    }
}

com.weibo.api.motan.core.ExecutorQueue#force

public boolean force(Runnable o) {
    if (threadPoolExecutor.isShutdown()) {
        throw new RejectedExecutionException("Executor not running, can't force a command into the queue");
    }
    // 再次放入任务队列
    return super.offer(o);
}

public boolean offer(Runnable o) {
    // 当前线程数
    int poolSize = threadPoolExecutor.getPoolSize();
    // 当前线程数已达到最大线程数，放入队列
    if (poolSize == threadPoolExecutor.getMaximumPoolSize()) {
        return super.offer(o);
    }
    // 提价的任务数小于等于核心线程，放入队列，让核心线程处理
    if (threadPoolExecutor.getSubmittedTasksCount() <= poolSize) {
        return super.offer(o);
    }
    // 当前线程数小于最大线程数，返回 false，即启动临时线程，处理任务，不再排队
    if (poolSize < threadPoolExecutor.getMaximumPoolSize()) {
        return false;
    }
    // 放入队列
    return super.offer(o);
}