java并发之线程池Executor知识要点 及 核心源码浅析

shutdown（）；

shutdownNow();

submit（）；

invokeAll（）;

invokeAny();

execute（）；

shutdown（）；

shutdownNow（）；

isTerminated（）;

publicThreadPoolExecutor(intcorePoolSize,intmaximumPoolSize,longkeepAliveTime,TimeUnitunit,BlockingQueue<Runnable>workQueue,ThreadFactorythreadFactory,RejectedExecutionHandlerhandler)

无界队列：LinkedBlockingQueue-可无限制的增加，它是newFixedThreadPool和newSingleThreadExecutor默认的任务队列

有界队列：ArrayBlockingQueue、有界的LinkedBlockingQueue、PriorityBlockingQueue-当队列满后，会执行饱和策略

同步移交：在线程之间移交的一种机制，当线程池是无界的或者是可以拒绝任务的就可以使用，newCachedThreadPool就使用的是SynchronousQueue

ThredPoolExecutor类中的4个静态内部类：

AbortPolicy：默认的Handler，该策略会抛出未检查的RegectedExecutionException

CallerRunsPolicy：调用者运行，将某些任务回退到调用者（一般为主线程），由于执行任务需要一定的时间，因此主线程至少在一段时间内不能提交任何任务，从而使得工作者线程有时间来处理正在执行的任务

DiscardPolicy：抛弃新提交的任务

DiscardOldestPolicy：将抛弃下一个将要执行的任务

/**

1.当线程池中的数目小于corePoolSize时，直接new一个Thread

2.当线程池数大于corePoolSize时，直接放入任务队列中

3.如果队列已经满了且线程池中线程数小于maximumPoolSize，则新建一个线程

*/

publicvoidexecute(Runnablecommand){

if(command==null)

thrownewNullPointerException();

intc=ctl.get();

//如果正在运行的线程数小于线程池预设的大小，就尝试addWorker(源码在其后)。若成功，直接返回。若添加失败(可能在添加过程中已达到预设的线程池的数目)，重新获取线程池正在运行的线程数

if(workerCountOf(c)<corePoolSize){

if(addWorker(command,true))

return;

c=ctl.get();

}

//若任务没有线程处理(当达到了线程池的预设大小corePoolSize)，就添加到任务队列。

if(isRunning(c)&&workQueue.offer(command)){

intrecheck=ctl.get();

if(!isRunning(recheck)&&remove(command))//若线程池在workeQueue.offer前发生了shutdown，就从任务队列中移除

reject(command);

elseif(workerCountOf(recheck)==0)//如果线程池在运行，并且没有可工作的线程，就直接创建一个

addWorker(null,false);

}

elseif(!addWorker(command,false))//如果任务队列已满，尝试创建一个新的Worker，若失败，说明线程池已经关闭或者饱和了

reject(command);//饱和策略问题

}

/**

@paramfirstTask:是新线程应该运行的第一个任务，Worker会被创建在线程池的数目比corePoolSize小的时候，或者在任务队列已经满的情况下，创建一个线程来代替已死的线程

@paramcore：当为true时，用corePoolSize作为边界，否则，用maximumPoolSize作为边界

*/

privatebooleanaddWorker(RunnablefirstTask,booleancore){

retry:

for(;;){

intc=ctl.get();

intrs=runStateOf(c);//获取线程池的状态

//如果线程池被shutdown了，一般直接返回false。但是排除任务队列不为空但Workers为空的情况，在这种情况下，会调用addWorker（null,false）来创建一个线程处理队列中的任务

if(rs>=SHUTDOWN&&

!(rs==SHUTDOWN&&

firstTask==null&&

!workQueue.isEmpty()))

returnfalse;

for(;;){//如果正在执行的线程数目大于线程池中预设的线程数，返回false

intwc=workerCountOf(c);

if(wc>=CAPACITY||wc>=(core?corePoolSize:maximumPoolSize))

returnfalse;

if(compareAndIncrementWorkerCount(c))

breakretry;

c=ctl.get();//Re-readctl

if(runStateOf(c)!=rs)

continueretry;

//elseCASfailedduetoworkerCountchange;retryinnerloop

}

}

booleanworkerStarted=false;

booleanworkerAdded=false;

Workerw=null;

try{

finalReentrantLockmainLock=this.mainLock;

w=newWorker(firstTask);//创建Worker：其中Thread通过调用ThreadFactory的newThread方法构建，所以在此处可以对创建的Thread进行额外的处理

finalThreadt=w.thread;

if(t!=null){

mainLock.lock();

try{

intc=ctl.get();

intrs=runStateOf(c);

//若线程池正在运行或者处于shutdown但是任务不为空，则把新建的worker添加在workers中

if(rs<SHUTDOWN||(rs==SHUTDOWN&&firstTask==null)){

if(t.isAlive())//precheckthattisstartable

thrownewIllegalThreadStateException();

workers.add(w);

ints=workers.size();//largePoolSize用于记录曾经出现过得最大的线程数

if(s>largestPoolSize)

largestPoolSize=s;

workerAdded=true;

}

}finally{

mainLock.unlock();

}

if(workerAdded){//启动任务的执行

t.start();

workerStarted=true;

}

}

}finally{

if(!workerStarted)

addWorkerFailed(w);//若启动失败，则从正在运行的工作集中移除

}

returnworkerStarted;

}

/**

当线程池被stop或者shutdown或创建线程失败时，则会调用这个方法

1.从workers中移除worker

2.把workerCount-1

3.尝试终止操作：当线程池的状态为shutdown、线程池的数目和任务队列都为空，或者线程池已经stop、线程池数目为0

*/

privatevoidaddWorkerFailed(Workerw){

finalReentrantLockmainLock=this.mainLock;

mainLock.lock();

try{

if(w!=null)

workers.remove(w);

decrementWorkerCount();

tryTerminate();

}finally{

mainLock.unlock();

}

}

finalvoidtryTerminate(){

for(;;){

intc=ctl.get();

if(isRunning(c)||

runStateAtLeast(c,TIDYING)||

(runStateOf(c)==SHUTDOWN&&!workQueue.isEmpty()))

return;

if(workerCountOf(c)!=0){//Eligibletoterminate

interruptIdleWorkers(ONLY_ONE);

return;}

finalReentrantLockmainLock=this.mainLock;

mainLock.lock();

try{

if(ctl.compareAndSet(c,ctlOf(TIDYING,0))){

try{

terminated();

}finally{

ctl.set(ctlOf(TERMINATED,0));

termination.signalAll();

}

return;

}

}finally{

mainLock.unlock();

}

//elseretryonfailedCAS

}

}

个人觉得还比较重要的一个内部类是Worker，继承了AbstractQueuedSynchronizer类，其中最主要的run方法

privatefinalclassWorker

extendsAbstractQueuedSynchronizer

implementsRunnable

{

Worker(RunnablefirstTask){

setState(-1);//inhibitinterruptsuntilrunWorker

this.firstTask=firstTask;

this.thread=getThreadFactory().newThread(this);

}

publicvoidrun(){

runWorker(this);

}

protectedbooleanisHeldExclusively(){

returngetState()!=0;

}

protectedbooleantryAcquire(intunused){

if(compareAndSetState(0,1)){

setExclusiveOwnerThread(Thread.currentThread());

returntrue;}

returnfalse;

}

protectedbooleantryRelease(intunused){

setExclusiveOwnerThread(null);

setState(0);

returntrue;}

//参数1就是把锁设置为了独占锁，在获取到一个任务后，准备执行前首先要获取这个锁。

publicvoidlock(){acquire(1);}

publicbooleantryLock(){returntryAcquire(1);}

publicvoidunlock(){release(1);}

publicbooleanisLocked(){returnisHeldExclusively();}

voidinterruptIfStarted(){

Threadt;

if(getState()>=0&&(t=thread)!=null&&!t.isInterrupted()){

try{

t.interrupt();

}catch(SecurityExceptionignore){

}

}

}

}

finalvoidrunWorker(Workerw){

Threadwt=Thread.currentThread();

//不从任务队列中获取第一个任务，而是执行刚提交的任务

Runnabletask=w.firstTask;

w.firstTask=null;

w.unlock();//allowinterrupts

booleancompletedAbruptly=true;

try{

while(task!=null||(task=getTask())!=null){

w.lock();

//如果线程池是STOP状态，则需要保证线程时被中断了

//ifnot,ensurethreadisnotinterrupted.This//requiresarecheckinsecondcasetodealwith//shutdownNowracewhileclearinginterrupt

if((runStateAtLeast(ctl.get(),STOP)||

(Thread.interrupted()&&

runStateAtLeast(ctl.get(),STOP)))&&

!wt.isInterrupted())

wt.interrupt();

try{

beforeExecute(wt,task);//执行前的勾子函数，可以通过重写的方式对这个函数进行扩展功能。当发生了异常时，是不会执行任务的，afterExecute也不会执行

Throwablethrown=null;

try{

task.run();//执行任务

}catch(RuntimeExceptionx){//当发生运行时异常或ERROR时，会原样抛出

thrown=x;throwx;

}catch(Errorx){

thrown=x;throwx;

}catch(Throwablex){//如果是一个Throwable，则会包装成一个Error抛出

thrown=x;thrownewError(x);

}finally{

afterExecute(task,thrown);//执行后的勾子函数，同样可以进行扩展

}

}finally{

task=null;

w.completedTasks++;

w.unlock();

}

}

completedAbruptly=false;

}finally{

processWorkerExit(w,completedAbruptly);

}

}

~

/**

从任务队列中获取一个任务，在以下情况会退出：

1.线程数大于maximumPoolSize，我也不知道为什么会大于？？创建的时候就会判断啊？？

2.线程池已经停止

3.线程池shutdown并且队列也为空

4.获取一个task，超时

@return：返回一个task，或者worker退出，workerCount-1

*/

privateRunnablegetTask(){

booleantimedOut=false;//Didthelastpoll()timeout?

retry:

for(;;){

intc=ctl.get();

intrs=runStateOf(c);

//如果当前线程池已经shutdown或者stop&&任务队列为空，则workerCount-1

if(rs>=SHUTDOWN&&(rs>=STOP||workQueue.isEmpty())){

decrementWorkerCount();

returnnull;}

booleantimed;//Areworkerssubjecttoculling?

for(;;){

intwc=workerCountOf(c);

timed=allowCoreThreadTimeOut||wc>corePoolSize;

if(wc<=maximumPoolSize&&!(timedOut&&timed))//若没有超时

break;

if(compareAndDecrementWorkerCount(c))

returnnull;

c=ctl.get();//Re-readctl

if(runStateOf(c)!=rs)

continueretry;

//elseCASfailedduetoworkerCountchange;retryinnerloop

}

try{

Runnabler=timed?

workQueue.poll(keepAliveTime,TimeUnit.NANOSECONDS):

workQueue.take();

if(r!=null)

returnr;

timedOut=true;//若没有获取到task，超时

}catch(InterruptedExceptionretry){//若被中断了，不能算作超时

timedOut=false;

}

}

}