深度解析 | 从原生到开源，如何扩展 Java 线程池？
开篇
Tomcat、Dubbo、Motan 的线程池并非是对 JDK 原生的简单封装，而是基于各自核心场景做了定制化源码扩展。三者均基于 JDK java.util.concurrent.ThreadPoolExecutor扩展，核心差异在队列实现和线程调度逻辑。默认你懂 JDK 线程池的核心流程（核心线程→队列→临时线程→拒绝策略）。
一、Tomcat（9.0.41）
Tomcat线程池 vs JDK原生线程池
Tomcat的核心线程池实现是org.apache.tomcat.util.threads.ThreadPoolExecutor（注意：不是JDK的java.util.concurrent.ThreadPoolExecutor），它继承自JDK原生类，但做了3个关键扩展，适配Web场景：
| 特性 | JDK原生ThreadPoolExecutor | Tomcat ThreadPoolExecutor |
|---|---|---|
| 核心线程预启动 | 默认不预启动，需手动调用prestartCoreThread() | 可配置prestartminSpareThreads，自动预启动核心线程 |
| 队列满后的处理逻辑 | 队列满→创建临时线程→触发拒绝策略 | 队列满→直接触发拒绝策略（Web场景优先拒绝，避免请求堆积） |
| 线程空闲销毁逻辑 | 核心线程默认永不销毁 | 可配置minSpareThreads，核心线程低于该值时不销毁 |
| 适用场景 | 通用业务场景（长任务/短任务均可） | Web场景（短连接、高并发、快速响应） |
10年经验注
Tomcat线程池的设计核心是“快进快出”——Web请求都是短任务，优先保证“连接能快速被处理”，而非像JDK线程池那样“尽量容纳任务”，这是调优的核心原则。
Tomcat线程池的核心实现类是ThreadPoolExecutor（Tomcat自定义），核心依赖TaskQueue（自定义队列），以下只拆Web场景最关键的逻辑：
线程池的创建入口
publicclassStandardThreadExecutorextendsLifecycleMBeanBaseimplementsExecutor,ResizableExecutor{//核心变量protectedString namePrefix ="tomcat-exec-";//最大线程数protectedint maxThreads =200;//核心线程数protectedint minSpareThreads =25;protectedint maxIdleTime =60000;//tomcat扩展的线程池protectedThreadPoolExecutor executor =null;//是否预先启动核心线程protectedboolean prestartminSpareThreads =false;//默认可堆积的最大任务数protectedint maxQueueSize =Integer.MAX_VALUE;//任务队列privateTaskQueue taskqueue =null;protectedvoidstartInternal()throwsLifecycleException{this.taskqueue =newTaskQueue(this.maxQueueSize);TaskThreadFactory tf =newTaskThreadFactory(this.namePrefix,this.daemon,this.getThreadPriority());//创建tomcat扩展的线程池，继承自java.util.concurrent.ThreadPoolExecutor this.executor =newThreadPoolExecutor(this.getMinSpareThreads(),this.getMaxThreads(),(long)this.maxIdleTime,TimeUnit.MILLISECONDS,this.taskqueue, tf);this.executor.setThreadRenewalDelay(this.threadRenewalDelay);if(this.prestartminSpareThreads){this.executor.prestartAllCoreThreads();}//线程池设置到任务队列this.taskqueue.setParent(this.executor);this.setState(LifecycleState.STARTING);}}线程池的执行入口
publicvoidexecute(Runnable command,long timeout,TimeUnit unit){//提交的任务数 +1this.submittedCount.incrementAndGet();try{//调用JDK线程池的execute方法。重点在 TaskQueue 的offer方法super.execute(command);}catch(RejectedExecutionException rx){if(!(super.getQueue()instanceofTaskQueue)){this.submittedCount.decrementAndGet();throw rx;}//任务队列已积满任务TaskQueue queue =(TaskQueue)super.getQueue();try{//再次尝试将任务放入队列if(!queue.force(command, timeout, unit)){this.submittedCount.decrementAndGet();thrownewRejectedExecutionException(sm.getString("threadPoolExecutor.queueFull"));}}catch(InterruptedException x){//提交的任务数 -1this.submittedCount.decrementAndGet();thrownewRejectedExecutionException(x);}}}核心队列：TaskQueue（Tomcat自定义的阻塞队列）。这是Tomcat线程池和JDK原生最核心的差异点，源码核心逻辑：
publicclassTaskQueueextendsLinkedBlockingQueue<Runnable>{privatetransientThreadPoolExecutor executor;publicbooleanforce(Runnable o,long timeout,TimeUnit unit)throwsInterruptedException{if(this.parent !=null&&!this.parent.isShutdown()){//此处调用父类LinkedBlockingQueue的offer方法returnsuper.offer(o, timeout, unit);}else{thrownewRejectedExecutionException(sm.getString("taskQueue.notRunning"));}}// 重写offer方法：Web场景下，队列满时直接返回false，触发拒绝策略publicbooleanoffer(Runnable o){if(this.parent ==null){returnsuper.offer(o);}elseif(this.parent.getPoolSize()==this.parent.getMaximumPoolSize()){//如果创建的线程数等于最大线程数，放入任务队列（此时只能排队等待处理）returnsuper.offer(o);}elseif(this.parent.getSubmittedCount()<=this.parent.getPoolSize()){// 提交的任务数小于核心线程数，放入队列（此时核心线程足以处理提交的任务，不需要创建临时线程）returnsuper.offer(o);}else{//创建的线程数小于最大线程数，返回false，此时提交的任务不会排队，而是创建临时线程；否则放入队列returnthis.parent.getPoolSize()<this.parent.getMaximumPoolSize()?false:super.offer(o);}}}源码解读
JDK原生线程池的逻辑是“核心线程→队列→临时线程→拒绝”，而Tomcat通过重写offer方法，改成了“核心线程→临时线程（直到maxThreads）→队列→拒绝”——这意味着Tomcat会优先创建线程处理请求，而非让请求排队，完美适配Web请求“短、快”的特性，减少请求排队等待时间。
核心扩展：预启动核心线程。Tomcat线程池支持配置prestartminSpareThreads="true"，源码层面会在初始化时调用。预启动核心线程能避免“首次请求创建线程的开销”——生产中建议开启，尤其是秒杀、活动等突发流量场景。
publicvoidprestartAllCoreThreads(){int n =prestartCoreThread();while(n <getCorePoolSize()){ n +=prestartCoreThread();}}二、Dubbo（3.3.0）
与Tomcat中线程池的实现类似
org.apache.dubbo.common.threadpool.support.eager.EagerThreadPool#getExecutor
publicExecutorgetExecutor(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 =newTaskQueue<Runnable>(queues <=0?1: queues);EagerThreadPoolExecutor executor =newEagerThreadPoolExecutor(cores, threads, alive,TimeUnit.MILLISECONDS, taskQueue,newNamedInternalThreadFactory(name,true),newAbortPolicyWithReport(name, url)); taskQueue.setExecutor(executor);return executor;}org.apache.dubbo.common.threadpool.support.eager.EagerThreadPoolExecutor#execute
publicvoidexecute(Runnable command){if(command ==null){thrownewNullPointerException();}// 提交的任务数 +1 submittedTaskCount.incrementAndGet();try{super.execute(command);}catch(RejectedExecutionException rx){// retry to offer the task into queue.finalTaskQueue queue =(TaskQueue)super.getQueue();try{//拒绝之后再次尝试放入队列if(!queue.retryOffer(command,0,TimeUnit.MILLISECONDS)){ submittedTaskCount.decrementAndGet();//任务数 -1thrownewRejectedExecutionException("Queue capacity is full.", rx);}}catch(InterruptedException x){ submittedTaskCount.decrementAndGet();thrownewRejectedExecutionException(x);}}catch(Throwable t){// decrease any way submittedTaskCount.decrementAndGet();throw t;}}org.apache.dubbo.common.threadpool.support.eager.TaskQueue#offer
publicbooleanoffer(Runnable runnable){if(executor ==null){thrownewRejectedExecutionException("The task queue does not have executor!");}//当前启动的线程数int currentPoolThreadSize = executor.getPoolSize();//提交的任务数小于核心线程数，说明核心线程足以处理,放入队列，让核心线程处理if(executor.getSubmittedTaskCount()< currentPoolThreadSize){returnsuper.offer(runnable);}//已创建的线程数小于最大线程数，放回false，不会将任务放入队列，而是创建临时线程if(currentPoolThreadSize < executor.getMaximumPoolSize()){returnfalse;}// 直接放入队列，排队returnsuper.offer(runnable);}默认拒绝策略
org.apache.dubbo.common.threadpool.support.AbortPolicyWithReport#rejectedExecution
publicvoidrejectedExecution(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);thrownewRejectedExecutionException(msg);}三、Motan（1.2.0）
com.weibo.api.motan.core.StandardThreadExecutor
继承JDK的线程池，设计思路借鉴于Tomcat
队列ExecutorQueue，继承LinkedTransferQueue，能保证更高性能，相比LinkedBlockingQueue有明显提升
publicStandardThreadExecutor(int coreThreads,int maxThreads,long keepAliveTime,TimeUnit unit,int queueCapacity,ThreadFactory threadFactory,RejectedExecutionHandler handler){//初始化super(coreThreads, maxThreads, keepAliveTime, unit,newExecutorQueue(),//任务队列，继承自LinkedTransferQueue threadFactory, handler);((ExecutorQueue)getQueue()).setStandardThreadExecutor(this);//统计提交的任务数 submittedTasksCount =newAtomicInteger(0);// 可以提交的最大任务数： 队列长度数 + 最大线程数 maxSubmittedTaskCount = queueCapacity + maxThreads;}publicvoidexecute(Runnable command){int count = submittedTasksCount.incrementAndGet();// 超过可以提交的任务数，进行 rejectif(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
publicbooleanforce(Runnable o){if(threadPoolExecutor.isShutdown()){thrownewRejectedExecutionException("Executor not running, can't force a command into the queue");}// 再次放入任务队列returnsuper.offer(o);}publicbooleanoffer(Runnable o){//当前线程数int poolSize = threadPoolExecutor.getPoolSize();// 当前线程数已达到最大线程数，放入队列if(poolSize == threadPoolExecutor.getMaximumPoolSize()){returnsuper.offer(o);}// 提价的任务数小于等于核心线程，放入队列，让核心线程处理if(threadPoolExecutor.getSubmittedTasksCount()<= poolSize){returnsuper.offer(o);}// 当前线程数小于最大线程数，返回false，即启动临时线程，处理任务，不再排队if(poolSize < threadPoolExecutor.getMaximumPoolSize()){returnfalse;}// 放入队列returnsuper.offer(o);}结尾
以上是我 10 年 Java 开发经验，从源码层面拆解的 Tomcat/Dubbo/Motan 线程池核心逻辑。
如果你还知道其他开源项目也对线程池进行了扩展，欢迎评论区聊聊。
