java 线程池 ThreadPoolExecutor 源码扩展支持先增加非核心线程处理任务后放任务队列

文章目录

继承体系
七个参数
RejectedExecutionHandler
- CallerRunsPolicy
- AbortPolicy
- DiscardPolicy
- DiscardOldestPolicy
- 自定义handler
扩展：核心线程满了以后，如何将任务先交由非核心线程处理
- 源码解析
- 扩展源码
- - 自定义 Queue，重写 offer 方法
  - 自定义线程池，继承 ThreadPoolExecutor，修改核心逻辑
- 总结
参考资料

阿里巴巴Java开发手册中强制规定，线程池不允许使用 Executors 去创建，而是通过 ThreadPoolExecutor 的方式，这样的处理方式让写的同学更加明确线程池的运行规则，规避资源耗尽的风险。

所以下面我们就康康 ThreadPoolExecutor 的7个参数。

继承体系

七个参数

    /*** Creates a new {@code ThreadPoolExecutor} with the given initial* parameters.** @param corePoolSize the number of threads to keep in the pool, even*        if they are idle, unless {@code allowCoreThreadTimeOut} is set* @param maximumPoolSize the maximum number of threads to allow in the*        pool* @param keepAliveTime when the number of threads is greater than*        the core, this is the maximum time that excess idle threads*        will wait for new tasks before terminating.* @param unit the time unit for the {@code keepAliveTime} argument* @param workQueue the queue to use for holding tasks before they are*        executed.  This queue will hold only the {@code Runnable}*        tasks submitted by the {@code execute} method.* @param threadFactory the factory to use when the executor*        creates a new thread* @param handler the handler to use when execution is blocked*        because the thread bounds and queue capacities are reached* @throws IllegalArgumentException if one of the following holds:<br>*         {@code corePoolSize < 0}<br>*         {@code keepAliveTime < 0}<br>*         {@code maximumPoolSize <= 0}<br>*         {@code maximumPoolSize < corePoolSize}* @throws NullPointerException if {@code workQueue}*         or {@code threadFactory} or {@code handler} is null*/public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler) {}

corePoolSize 核心线程数，不会被回收，除非设置了 allowCoreThreadTimeOut
maximumPoolSize 线程池中允许最大的线程数量。
- corePoolSize和maximumPoolSize的关系类似于工厂的编制员工和临时工。corePoolSize是编制员工的数量，（maximumPoolSize - corePoolSize）的数量是工厂订单旺季时，可以招的临时工的数量。等到工厂淡季的时候，就把这些临时工解散，但是编制员工淡季也不解散。
keepAliveTime 空闲线程被终止前的等待时间
unit 等待时间的单位
workQueue 任务队列。线程池提交任务时，先交由空闲的核心线程处理，没有空闲的核心线程以后就放到阻塞队列里。阻塞队列也满了就创建非核心线程来处理。但是非核心线程的数量加核心线程的数量不能超过 maximumPoolSize。
- ArrayBlockingQueue
- LinkedBlockingQueue
- SynchronousQueue
- PriorityBlockQueue
threadFactory 创建线程的工厂
handler 拒绝策略。当线程池中线程数量达到maximumPoolSize以后采取的策略，是丢掉还是让线程去执行，也可以自定义现实。

RejectedExecutionHandler

默认实现的拒绝策略

CallerRunsPolicy

提交任务的线程去处理

    /*** A handler for rejected tasks that runs the rejected task* directly in the calling thread of the {@code execute} method,* unless the executor has been shut down, in which case the task* is discarded.*/public static class CallerRunsPolicy implements RejectedExecutionHandler {}

这个有点不好理解，举个栗子，比如现在线程池满了，main方法中又来了一个线程，那么这个任务就交给main线程去处理掉。

AbortPolicy

抛个异常出去

    /*** A handler for rejected tasks that throws a* {@code RejectedExecutionException}.*/public static class AbortPolicy implements RejectedExecutionHandler {}

DiscardPolicy

默默丢掉新来的任务

    /*** A handler for rejected tasks that silently discards the rejected task.*/public static class DiscardPolicy implements RejectedExecutionHandler {}

DiscardOldestPolicy

丢掉workQueue里排队最久还没被处理掉的任务，然后重复去excute这个任务。除非，excutor被shut down了，那这个任务就被丢弃了。

    /*** A handler for rejected tasks that discards the oldest unhandled* request and then retries {@code execute}, unless the executor* is shut down, in which case the task is discarded.*/public static class DiscardOldestPolicy implements RejectedExecutionHandler {}

自定义handler

import java.util.concurrent.*;public class T14_MyRejectedHandler {public static void main(String[] args) {ExecutorService service = new ThreadPoolExecutor(4, 4,0, TimeUnit.SECONDS, new ArrayBlockingQueue<>(6),Executors.defaultThreadFactory(),new MyHandler());}static class MyHandler implements RejectedExecutionHandler {@Overridepublic void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {//自定义处理方式，可以先打个日志 xxx 被我拒绝了 然后我把它放到了 redis 里 ...//log("r rejected")//save r in kafka、mysql、redis}}
}

扩展：核心线程满了以后，如何将任务先交由非核心线程处理

上文我们介绍线程池的核心参数 workQueue 时讲到，线程池提交任务时，先交由空闲的核心线程处理，没有空闲的核心线程以后就放到阻塞队列里。阻塞队列也满了就创建非核心线程来处理。

但是有没有一种可能，没有空闲的核心线程以后，先创建非核心线程，直到核心线程数 + 非核心线程达到 maximumPoolSize 再将任务放到队列里。

答案是有的，在 Dubbo 源码中就有现成的方案。Github issue ：Extension: Eager Thread Pool #1568

看最后一句，Reference: Tomcat’s thread pool design 其实他也是借鉴了 Tomcat 的设计。

源码解析

先一起看下扩展之前，ThreadPoolExecutor 的 execute() 方法的逻辑，此处保留了源码中的注释。

    /*** Executes the given task sometime in the future.  The task* may execute in a new thread or in an existing pooled thread.** If the task cannot be submitted for execution, either because this* executor has been shutdown or because its capacity has been reached,* the task is handled by the current {@link RejectedExecutionHandler}.** @param command the task to execute* @throws RejectedExecutionException at discretion of*         {@code RejectedExecutionHandler}, if the task*         cannot be accepted for execution* @throws NullPointerException if {@code command} is null*/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();if (workerCountOf(c) < corePoolSize) {// 创建线程if (addWorker(command, true))return;c = ctl.get();}// 线程数量达到了 corePoolSize，新任务放入队列if (isRunning(c) && workQueue.offer(command)) {int recheck = ctl.get();if (! isRunning(recheck) && remove(command))reject(command);else if (workerCountOf(recheck) == 0)addWorker(null, false);}// 入队失败，创建非核心线程else if (!addWorker(command, false))// 创建新线程失败执行拒绝策略reject(command);}

扩展源码

自定义 Queue，重写 offer 方法

package eagerthreadpool;import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;public class TaskQueue<R extends Runnable> extends LinkedBlockingQueue<Runnable> {private static final long serialVersionUID = -2635853580887179627L;// 自定义的线程池类，继承自ThreadPoolExecutorprivate EagerThreadPoolExecutor executor;public TaskQueue(int capacity) {super(capacity);}public void setExecutor(EagerThreadPoolExecutor exec) {executor = exec;}// offer方法的含义是：将任务提交到队列中，返回值为true/false，分别代表提交成功/提交失败@Overridepublic 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);}// return false to let executor create new worker.if (currentPoolThreadSize < executor.getMaximumPoolSize()) {// 重点： 当前线程数小于 最大线程数 ，返回false，暗含入队失败，让线程池去创建新的线程return false;}// 重点: 代码运行到此处，说明当前线程数 >= 最大线程数，需要真正的提交到队列中return super.offer(runnable);}public boolean retryOffer(Runnable o, long timeout, TimeUnit unit) throws InterruptedException {if (executor.isShutdown()) {throw new RejectedExecutionException("Executor is shutdown!");}return super.offer(o, timeout, unit);}
}

自定义线程池，继承 ThreadPoolExecutor，修改核心逻辑

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;public class EagerThreadPoolExecutor extends ThreadPoolExecutor {/*** 定义一个成员变量，用于记录当前线程池中已提交的任务数量，在队列的 offer() 方法中要用*/private final AtomicInteger submittedTaskCount = new AtomicInteger(0);public EagerThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit, TaskQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler) {super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);}public int getSubmittedTaskCount() {return submittedTaskCount.get();}@Overrideprotected void afterExecute(Runnable r, Throwable t) {// ThreadPoolExecutor的勾子方法，在task执行完后需要将池中已提交的任务数 - 1submittedTaskCount.decrementAndGet();}@Overridepublic void execute(Runnable command) {if (command == null) {throw new NullPointerException();}// do not increment in method beforeExecute!// 将池中已提交的任务数 + 1submittedTaskCount.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();throw new RejectedExecutionException("Queue capacity is full.", rx);}} catch (InterruptedException x) {submittedTaskCount.decrementAndGet();throw new RejectedExecutionException(x);}} catch (Throwable t) {// decrease any waysubmittedTaskCount.decrementAndGet();throw t;}}
}

总结

自定义的 EagerThreadPoolExecutor 依赖自定义的 TaskQueue 的 offer 返回值来决定是否创建更多的线程，达到先判断 maximumPoolSize 再判断队列的目的。

参考资料

雄哥：Java如何让线程池满后再放队列
ThreadPoolExecutor 源码解析