java-源码解读-线程池实现原理-0

线程池初识

我们一般这样使用线程池：

public class ExecutorServiceDemo {
public static void main(String[] args) {
// 创建一个线程池对象，控制要创建几个线程对象。
ExecutorService pool = Executors.newFixedThreadPool(2);
// 可以执行Runnable对象或者Callable对象代表的线程
List<FutureTask> futures=new ArrayList<FutureTask>();
for(int i=0;i<n;i++){
FutureTask future=pool.submit(new MyRunnable());
futures.add(future);
}
//线程池不接收新的任务
pool.shutdown();
//做一些别的事情
//...

//获取结果
//future.get()
}
}

首先获取一个线程池，获取的方法可以用Executors工具类，也可以是别的方法，然后向线程池中提交任务，然后获取结果。

深入线程池

线程的状态和工作线程个数

线程池用ctl变量表示线程池的状态和工作线程数

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

从这里也可以看出线程池的初始状态为RUNNING  

ctl的高三位表示线程池的状态，低29位表示工作线程的个数。

private static final int COUNT_BITS = Integer.SIZE - 3; //count为低29位表示

线程池的状态

// runState is stored in the high-order bits  线程池用ctl高三位表示状态
private static final int RUNNING    = -1 << COUNT_BITS;

        转成二进制为：RUNNING     = 111 xxxxxxxxxxxxxxxxxxxxxxxxxxxxx

private static final int SHUTDOWN   =  0 << COUNT_BITS;

          转成二进制为：SHUTDOWN   =  000 xxxxxxxxxxxxxxxxxxxxxxxxxxxxx

private static final int STOP   =  1 << COUNT_BITS;   

          转成二进制为：STOP  =  001 xxxxxxxxxxxxxxxxxxxxxxxxxxxxx

private static final int TIDYING    =  2 << COUNT_BITS;

          转成二进制为：TIDYING    = 010 xxxxxxxxxxxxxxxxxxxxxxxxxxxxx

private static final int TERMINATED =  3 << COUNT_BITS;          转成二进制为：TERMINATED = 011 xxxxxxxxxxxxxxxxxxxxxxxxxxxxx

线程池的容量

private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

          转成二进制为：CAPACITY   = 000 11111111111111111111111111111 

     
构建线程池的成员

/**线程池工厂，用于创建线程**/

private volatile ThreadFactory threadFactory;

/**拒绝策略**/

private volatile RejectedExecutionHandler handler;

/**线程空闲时存活时间**/

private volatile long keepAliveTime;

/**是否允许线程空闲**/

private volatile boolean allowCoreThreadTimeOut;

/**core pool size**/

private volatile int corePoolSize;

/**max pool size**/

private volatile int maximumPoolSize;

/**默认的拒绝策略 为放弃**/

private static final RejectedExecutionHandler defaultHandler =  new AbortPolicy();

构建线程池

用线程池工具类Executors

ExecutorService servcie=Executors.newFixedThreadPool(..),

ExecutorService servcie=Executors.newCachedThreadPool(..)

...

调用构造方法构造线程池

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的构造方法

提交任务的时候发生了什么

不管是用ThreadPoolExecutor.execute()提交，还是用ExecutorService的submit方法提交，最终都会调用execute方法。

(execute方法提交和subit提交有何不同呢，请参考我的另一遍博客：java-源码解读-线程池提交之execute和submit有何不同)

public <T> Future<T> submit(Callable<T> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task);
execute(ftask);  //此处还是调用了execute方法提交
return ftask;
}

开始提交任务

/**execute方法提交任务**/
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
//获取活动线程数和线程池状态，这两步是一个原子操作，通过获取ctl同时获取（高3位是状态借低29位是活动线程数），
int c = ctl.get();
/**A:如果活动线程数小于corePoolSize就创建一个新的线程来执行任务**/

if (workerCountOf(c) < corePoolSize) {
/**addWorker方法创建一个新的线程**/
if (addWorker(command, true))
return;
/**addWorker有可能会失败，可能中因为线程池关闭了。。。这时需要重新获取ctl**/
c = ctl.get();
}
/**B:如果前面addWorker失败了，线程池没关闭，就把任务加入到工作队列中**/
if (isRunning(c) && workQueue.offer(command)) {
/**重新获取ctl,线程池的状态有可能又变了，或者有一个线程已执行完**/
int recheck = ctl.get();
/**如果线程池已经关闭了，就执行拒绝策略**/
if (! isRunning(recheck) && remove(command))
reject(command);
else if (workerCountOf(recheck) == 0) /**没有工作线程**/
/**如果此时线程池已经关闭了，但已经没有工作线程了，
就提交一个空任务，然后创建一个线程执行队列的任务，
addWorker方法中会判断线程池的状态，
添加一个null任务是因为SHUTDOWN状态下，线程池不再接受新任务**/
addWorker(null, false);
}
/**如果AB两步都失败了,即此时工作线程数>corePoolSize,然后加入队列也失败了，此时就只有增加一个新的线程来执行了**/
/**这个时候有两种情况：
1.非RUNNING状态拒绝新的任务
2.队列满了workCount < maximumPoolSize（addWorker方法中会有判断），注：此时addWorker的参数为false
**/
else if (!addWorker(command, false))
/**如果还失败就执行拒绝策略**/
reject(command);
}

从上面可以看出：

A:如果活动线程数小于corePoolSize就创建一个新的线程来执行任务

B:如果addWorker失败了,大部分原因是workerCountOf(c)>=corePoolSize，且线程池没关闭，就把任务加入到工作队列中
C:如果AB都失败，即 maxpoolsize>工作线程>corepoolSize,就只有创建一个新的线程来运行任务

工作线程是怎样运行起来

/**创建新的线程(Worker)执行Runnable,只有添加空任务时 core=false**/
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);

// Check if queue empty only if necessary.
/**状态检查，状态为SHUTDOWN的就不能添加新任务了
但此处为first==null(空)任务开了绿灯，
当rs>=SHUTDOWN,firtTask==null workQueue.isEmpty==false时,不会return false ,而是继续执行下面的逻辑
**/
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;

for (;;) {
//判断当前的工作线程数
int wc = workerCountOf(c);
//只有添加空任务时 core=false
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
//CAS操作使workcount+1,成功了下跳出retry,失败了就继续retry
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get();  // Re-read ctl
//如果CAS失败了，可能线程池的状态又变了，继续retry
if (runStateOf(c) != rs)
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
}

boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
//创建了Worker,Worker中有Thread 成员变量，用来运行任务
w = new Worker(firstTask);
//t:Worker 中的线程
final Thread t = w.thread;
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//再一次检查状态
int rs = runStateOf(ctl.get());
//此处又为空任务亮了绿亮
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
//防止任务重复运行
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
//workers用来存放工作的线程
workers.add(w);
int s = workers.size();
//largestPoolSize只是作了记录而以
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
//工作线程启动了，这里启动了Worker，下面会看Worker的run方法里都有什么
t.start();
//标识任务已经起动了
workerStarted = true;
}
}
} finally {
//任务没有启动成功，有可能是线程池关闭了...要回滚，addWorkerFailed作一些回滚操作
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}

提交的任务以及队列中的任务何时调用

    

下面看最重要的Worker
 //首行worker是一个线程
private final class Worker extends AbstractQueuedSynchronizer implements Runnable{
//用来运行任务的线程
final Thread thread;
//要运行的任务
Runnable firstTask;
//已完成的任务
volatile long completedTasks;
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
//在这里运行任务
public void run() {
runWorker(this);
}

//在这里运行任务
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
//空任务是不会运行的，getTask()方法会用到线程池的两个变量keepAliveTime，allowCoreThreadTimeOut，后面讲
//取一个任务运行，这个任务有可能是刚刚提交的任务，也有可能是从队列中取的任务
//但是刚提交的任务会优先运行，如果没有提交的任务（或提交的为空任务），才调用getTask()从队列中取任务
while (task != null || (task = getTask()) != null) {
w.lock();
//判断任务有没有被取消
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
//这是一个空的回调方法
beforeExecute(wt, task);
Throwable thrown = null;
try {
//任务正式运行了
task.run();
} catch (RuntimeException x) {
thrown = x; throw x;
} catch (Error x) {
thrown = x; throw x;
} catch (Throwable x) {
thrown = x; throw new Error(x);
} finally {
//这是一个空的回调方法
afterExecute(task, thrown);
}
} finally {
task = null;
w.completedTasks++;
w.unlock();
}
}
//没有取到任务
completedAbruptly = false;
} finally {
//没有取到任务，worker要退出了
processWorkerExit(w, completedAbruptly);
}
}

    //getTask()方法从列队中取任务，用到线程池的两个变量keepAliveTime，allowCoreThreadTimeOut
private Runnable getTask() {
boolean timedOut = false;
//循环取，有可能会超时
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);

// 线程池状态检查
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
//
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}

try {
//在keepAliveTime时间内取到一个任务就返回
//从队列中取任务
               Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
}

从上面可以看出Worker的run方法一直在取任务，然后运行任务。如果在keepAliveTime时间内还没有取到任务，线程可能就要退出了