JAVA 之ThreadPoolExecutor執行緒池原理及其execute方法實例詳解

public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue workQueue)

// Executors#newFixedThreadPool
public static ExecutorService newFixedThreadPool(int nThreads) {
 return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue());
}

//Executors# newSingleThreadExecutor
public static ExecutorService newSingleThreadExecutor() {
 return new FinalizableDelegateExecutorService(new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue()));
}

//Executors#newCachedThreadPool
public static ExecutorService newCachedThreadPool() {
　　return new ThreadPooExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue());
}

//Executor#execute
public interface Executor {
 void execute(Runnable command);
}

//ThreadPoolExecutor#execute
public void execute(Runnable command) {
　　if (command == null) 
  throw new NullPointerException();
　　int c = ctl.get(); 
 …
}

//ThreadPoolExecutor
private static final int COUNT_BITS = Integer.SIZE – 3;  //32-3=29，线程数量所占位数
private static final int CAPACITY = (1 << COUNT_BITS) – 1; //低29位表示最大线程数，229-1
//五种线程池状态
private static final int RUNNING = -1 << COUNT_BITS; /int型变量高3位（含符号位）101表RUNING
private static final int SHUTDOWN = 0 << COUNT_BITS; //高3位000
private static final int STOP = 1 << COUNT_BITS; //高3位001
private static final int TIDYING = 2 << COUNT_BITS; //高3位010
private static final int TERMINATED = 3 << COUNT_BITS; //高3位011

//ThreadPoolExecutor#execute
public void execute(Runnable command) {
 if (command == null) 
  throw new NullPointerException();
　　 int c = ctl.get(); //由它可以获取到当前有效的线程数和线程池的状态
/*1.获取当前正在运行线程数是否小于核心线程池，是则新创建一个线程执行任务，否则将任务放到任务队列中*/
 if (workerCountOf(c) < corePoolSize){
  if (addWorker(command, tre))  //在addWorker中创建工作线程执行任务
   return ;
  c = ctl.get();
 }
/*2.当前核心线程池中全部线程都在运行workerCountOf(c) >= corePoolSize，所以此时将线程放到任务队列中*/
 if (isRunning(c) && workQueue.offer(command)) { //线程池是否处于运行状态，且是否任务插入任务队列成功
  int recheck = ctl.get();
　　　　 if (!isRunning(recheck) && remove(command))  //线程池是否处于运行状态，如果不是则使刚刚的任务出队
　　　　　　 reject(command); //抛出RejectedExceptionException异常
　　　　 else if (workerCountOf(recheck) == 0)
　　　　　　 addWorker(null, false);
　　}
/*3.插入队列不成功，且当前线程数数量小于最大线程池数量，此时则创建新线程执行任务，创建失败抛出异常*/
　　else if (!addWorker(command, false)){
　　　　reject(command); //抛出RejectedExceptionException异常
　　}
}

//ThreadPoolExecutor#addWorker
private boolean addWorker(Runnable firstTask, boolean core) {
/*首先会再次检查线程池是否处于运行状态，核心线程池中是否还有空闲线程，都满足条件过后则会调用compareAndIncrementWorkerCount先将正在运行的线程数+1，数量自增成功则跳出循环，自增失败则继续从头继续循环*/
　　...
　　if (compareAndIncrementWorkerCount(c))
　　　　break retry;
　　...
/*正在运行的线程数自增成功后则将线程封装成工作线程Worker*/
　　boolean workerStarted = false;
　　boolean workerAdded = false;
　　Worker w = null;
　　try {
　　　　final ReentrantLock mainLock = this.mainLock;  //全局锁
　　　　w = new Woker(firstTask);  //将线程封装为Worker工作线程
　　　　final Thread t = w.thread;
　　　　if (t != null) {
　　　　　　mainLock.lock(); //获取全局锁
/*当持有了全局锁的时候，还需要再次检查线程池的运行状态等*/
　　　　　　try {
　　　　　　　　int c = clt.get();
　　　　　　　　int rs = runStateOf(c);  //线程池运行状态
　　　　　　　　if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)){  //线程池处于运行状态，或者线程池关闭且任务线程为空
　　　　　　　　　　if (t.isAlive()) //线程处于活跃状态，即线程已经开始执行或者还未死亡，正确的应线程在这里应该是还未开始执行的
　　　　　　　　　　　　throw new IllegalThreadStateException();
　　　　　　　　　　workers.add(w); //private final HashSet wokers = new HashSet();包含线程池中所有的工作线程，只有在获取了全局的时候才能访问它。将新构造的工作线程加入到工作线程集合中
　　　　　　　　　　int s = worker.size(); //工作线程数量
　　　　　　　　　　if (s > largestPoolSize)
　　　　　　　　　　　　largestPoolSize = s;
　　　　　　　　　　workerAdded = true; //新构造的工作线程加入成功
　　　　　　　　}
　　　　　　} finally {
　　　　　　　　mainLock.unlock();
　　　　　　}
　 　　　　if (workerAdded) {
　　　　　　　　t.start(); //在被构造为Worker工作线程，且被加入到工作线程集合中后，执行线程任务，注意这里的start实际上执行Worker中run方法，所以接下来分析Worker的run方法
　　　　　　　　workerStarted = true;
　　　　　　}
　　　　}
　　} finally {
　　　　if (!workerStarted) //未能成功创建执行工作线程
　　　　　　addWorkerFailed(w); //在启动工作线程失败后，将工作线程从集合中移除
　　}
　　return workerStarted;
}

//ThreadPoolExecutor$Worker，它继承了AQS，同时实现了Runnable，所以它具备了这两者的所有特性
private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
　　final Thread thread;
　　Runnable firstTask;
　　public Worker(Runnable firstTask) {
　　　　setState(-1); //设置AQS的同步状态为-1，禁止中断，直到调用runWorker
　　　　this.firstTask = firstTask;
　　　　this.thread = getThreadFactory().newThread(this); //通过线程工厂来创建一个线程，将自身作为Runnable传递传递
　　}
　　public void run() {
　　　　runWorker(this); //运行工作线程
　　}
}