ThreadPoolExecutor类
java.uitl.concurrent.ThreadPoolExecutor类是线程池中最核心的一个类，我们平常使用的线程池都是通过这个类来实现的。
先来看看ThreadPoolExecutor类的继承关系以及其构造方法：
public class ThreadPoolExecutor extends AbstractExecutorService {
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), defaultHandler);
}
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
threadFactory, defaultHandler);
}
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
RejectedExecutionHandler handler) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), handler);
}
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
if (corePoolSize < 0 ||
maximumPoolSize <= 0 ||
maximumPoolSize < corePoolSize ||
keepAliveTime < 0)
throw new IllegalArgumentException();
if (workQueue == null || threadFactory == null || handler == null)
throw new NullPointerException();
this.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}
从上面的代码得知：
首先分析它的继承关系
ThreadPoolExecutor继承了AbstractExecutorService，
看下他们的继承关系：
Executor接口
public interface Executor {
/**
* Executes the given command at some time in the future. The command
* may execute in a new thread, in a pooled thread, or in the calling
* thread, at the discretion of the {@code Executor} implementation.
*
* @param command the runnable task
* @throws RejectedExecutionException if this task cannot be
* accepted for execution
* @throws NullPointerException if command is null
*/
void execute(Runnable command);
就只有一个抽象方法
ExecutorService接口继承了Executor接口，并加了很多自己的抽象方法
AbstractExecutorService抽象类实现了ExecutorService 接口
public abstract class AbstractExecutorService implements ExecutorService {
public <T> Future<T> submit(Runnable task, T result) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task, result);
execute(ftask);
return ftask;
}
/**
* @throws RejectedExecutionException {@inheritDoc}
* @throws NullPointerException {@inheritDoc}
*/
public <T> Future<T> submit(Callable<T> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task);
execute(ftask);
return ftask;
}
......
......其他代码
从上面的submit方法中可以得知：它最终调用的还是 execute(xxx);方法。而ThreadPoolExecutor则是直接调用execute(xxx);方法
继承关系分析完了，再来看看它的构造函数
ThreadPoolExecutor构造函数详解
从上面的构造函数代码中可以看出前面三个构造器都是调用的第四个构造器进行的初始化工作。下面来具体讲解构造函数中的几个参数作用
①corePoolSize：核心池的大小，这个参数跟后面讲述的线程池的实现原理有非常大的关系。在创建了线程池后，默认情况下，线程池中并没有任何线程，而是等待有任务到来才创建线程去执行任务，除非调用了prestartAllCoreThreads()或者prestartCoreThread()方法，从这2个方法的名字就可以看出，是预创建线程的意思，即在没有任务到来之前就创建corePoolSize个线程或者一个线程。默认情况下，在创建了线程池后，线程池中的线程数为0，当有任务来之后，就会创建一个线程去执行任务，当线程池中的线程数目达到corePoolSize后，就会把到达的任务放到缓存队列当中；
②maximumPoolSize：线程池最大线程数，这个参数也是一个非常重要的参数，它表示在线程池中最多能创建多少个线程；
③keepAliveTime：表示线程没有任务执行时最多保持多久时间会终止。默认情况下，只有当线程池中的线程数大于corePoolSize时，keepAliveTime才会起作用，直到线程池中的线程数不大于corePoolSize，即当线程池中的线程数大于corePoolSize时，如果一个线程空闲的时间达到keepAliveTime，则会终止，直到线程池中的线程数不超过corePoolSize。但是如果调用了allowCoreThreadTimeOut(boolean)方法，在线程池中的线程数不大于corePoolSize时，keepAliveTime参数也会起作用，直到线程池中的线程数为0；
④unit：参数keepAliveTime的时间单位，有7种取值，在TimeUnit类中有7种静态属性：
TimeUnit.DAYS; //天
TimeUnit.HOURS; //小时
TimeUnit.MINUTES; //分钟
TimeUnit.SECONDS; //秒
TimeUnit.MILLISECONDS; //毫秒
TimeUnit.MICROSECONDS; //微妙
TimeUnit.NANOSECONDS; //纳秒
⑤workQueue：一个阻塞队列，用来存储等待执行的任务，这个参数的选择也很重要，会对线程池的运行过程产生重大影响，一般来说，这里的阻塞队列有以下几种选择：
ArrayBlockingQueue;
LinkedBlockingQueue;
SynchronousQueue;
ArrayBlockingQueue和PriorityBlockingQueue使用较少，一般使用LinkedBlockingQueue和Synchronous。线程池的排队策略与BlockingQueue有关。
⑥threadFactory：线程工厂，主要用来创建线程；
handler：表示当拒绝处理任务时的策略，有以下四种取值：
ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。
ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常。
ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程）
ThreadPoolExecutor.CallerRunsPolicy：由调用线程处理该任务
参考[深入理解Java之线程池]
例子
这段代码如何执行的呢？
当执行
ExecutorService executor = Executors.newFixedThreadPool(1);
将调用下面的代码创建一个FixedThreadPool线程池
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
上面代码可以看出创建的线程池就是ThreadPoolExecutor，下面来分析它的具体运行流程：
首先看下execute()方法
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();
}
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);
}
上面的注释已经写的很全了，要是英文好的话可以直接看注释，如果觉得看注释很憋屈的话，可以看我分析。
①先判断传入的任务是否为空，空的话就直接抛出空指针异常了
if (command == null)
throw new NullPointerException();
，否则执行②
②代码如下：
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
首先来看第一个 if (workerCountOf(c) < corePoolSize) 语句：通过workerCountOf(c)；方法获取当前线程池的线程数
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
int c = ctl.get();
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
private static int workerCountOf(int c) { return c & CAPACITY; }
然后判断当前线程池的线程数是否大于核心线程池大小，是的话跳到③，否则执行
if (addWorker(command, true))
语句：添加任务到工作队列中，添加成功，结束运行流程。添加不成功，更新变量c（这里为啥要更新变量c呢？我觉得主要原因是：在多线程的情况下，添加当前任务时，其他任务已经被成功添加工作队列中了，可能线程池线程数量已经超出核心线程数了，从而导致当前任务addWorker(command, true)失败，或者其他线程执行了shutdown语句，所以这里需要给c变量重新赋值），跳到③
③执行到这里，说明当前线程池的线程数量已经大于核心线程数的大小了。
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);
}
首先判断当前线程池是否处于运行（running）状态，也就是小于0
private static final int RUNNING = -1 << COUNT_BITS;
private static final int SHUTDOWN = 0 << COUNT_BITS;
private static final int STOP = 1 << COUNT_BITS;
private static final int TIDYING = 2 << COUNT_BITS;
private static final int TERMINATED = 3 << COUNT_BITS;
private static boolean isRunning(int c) {
return c < SHUTDOWN;
}
是的话，就把任务添加到缓存队列中；
workQueue.offer(command)
添加任务到缓存队列成功后，执行下面的代码：
int recheck = ctl.get();
if (! isRunning(recheck) && remove(command))
reject(command);
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
刚开始不太理解：为啥把任务添加到缓存队列中后，还要执行这些代码？后来仔细看了源码才知道，这样写的目的依我看至少有两个目的：一：防止在将此任务添加进任务缓存队列的同时其他线程突然调用shutdown或者shutdownNow方法关闭了线程池
二：确保加入到缓存队列中的任务能够被执行完。
下面对这些代码进行分析：
首先判断当前线程池是否处于运行状态，
这句判断是为了防止在将此任务添加进任务缓存队列的同时其他线程突然调用shutdown或者shutdownNow方法关闭了线程池的一种应急措施
不是的话，说明线程池被关闭了。就把前面已经加到缓存队列中的任务移除掉，移除成功后，然后执行
reject(command);
如果当前线程池处于运行时状态，或者任务移除出缓存队列不成功，并且线程池中的线程数为0
workerCountOf(recheck) == 0
那么调用
addWorker(null, false);
方法创建一个线程来执行前面加入到缓存队列中的任务，确保加入到缓存队列中的任务能够被执行完
④如果当前线程池不处于运行时状态 【这里有点疑惑？？不过调用addWorker(command, false)这个方法时还是会判断是否处于运行状态，不是的话直接返回false了，所以添加任务到工作队列中的操作也是失败的】
或者缓存队列已经满了
workQueue.offer(command))//这句代码返回false
那么尝试去添加任务到工作队列中，
else if (!addWorker(command, false))
reject(command);
直到加入的任务数大于或者等于最大线程数，也就是线程池中的线程数>=maximumPoolSize 。添加失败，说明线程池中的线程数>=maximumPoolSize ，则调用
reject(command);
线程池的大概运行流程已经分析完，下篇文章将分析上面提到的
addWorker(command, false)；
workQueue.offer(command)；
reject(command);
这几个方法。
网友评论