Java concurrency线程池之线程池原理(二)_动力节点Java学院整理

2024-02-19 22:25:03 155

线程池示例

在分析线程池之前，先看一个简单的线程池示例。

importjava.util.concurrent.Executors;
importjava.util.concurrent.ExecutorService;

publicclassThreadPoolDemo1{

publicstaticvoidmain(String[]args){
//创建一个可重用固定线程数的线程池
ExecutorServicepool=Executors.newFixedThreadPool(2);
//创建实现了Runnable接口对象，Thread对象当然也实现了Runnable接口
Threadta=newMyThread();
Threadtb=newMyThread();
Threadtc=newMyThread();
Threadtd=newMyThread();
Threadte=newMyThread();
//将线程放入池中进行执行
pool.execute(ta);
pool.execute(tb);
pool.execute(tc);
pool.execute(td);
pool.execute(te);
//关闭线程池
pool.shutdown();
}
}

classMyThreadextendsThread{

@Override
publicvoidrun(){
System.out.println(Thread.currentThread().getName()+"isrunning.");
}
}

运行结果：

pool-1-thread-1isrunning.
pool-1-thread-2isrunning.
pool-1-thread-1isrunning.
pool-1-thread-2isrunning.
pool-1-thread-1isrunning.

示例中，包括了线程池的创建，将任务添加到线程池中，关闭线程池这3个主要的步骤。稍后，我们会从这3个方面来分析ThreadPoolExecutor。

线程池源码分析

(一)创建“线程池”

下面以newFixedThreadPool()介绍线程池的创建过程。

1.newFixedThreadPool()

newFixedThreadPool()在Executors.java中定义，源码如下：

publicstaticExecutorServicenewFixedThreadPool(intnThreads){
returnnewThreadPoolExecutor(nThreads,nThreads,
0L,TimeUnit.MILLISECONDS,
newLinkedBlockingQueue());
}

说明：newFixedThreadPool(intnThreads)的作用是创建一个线程池，线程池的容量是nThreads。
newFixedThreadPool()在调用ThreadPoolExecutor()时，会传递一个LinkedBlockingQueue()对象，而LinkedBlockingQueue是单向链表实现的阻塞队列。在线程池中，就是通过该阻塞队列来实现"当线程池中任务数量超过允许的任务数量时，部分任务会阻塞等待"。
关于LinkedBlockingQueue的实现细节，读者可以参考"Java多线程系列--“JUC集合”08之LinkedBlockingQueue"。

2.ThreadPoolExecutor()

ThreadPoolExecutor()在ThreadPoolExecutor.java中定义，源码如下：

publicThreadPoolExecutor(intcorePoolSize,
intmaximumPoolSize,
longkeepAliveTime,
TimeUnitunit,
BlockingQueueworkQueue){
this(corePoolSize,maximumPoolSize,keepAliveTime,unit,workQueue,
Executors.defaultThreadFactory(),defaultHandler);
}

说明：该函数实际上是调用ThreadPoolExecutor的另外一个构造函数。该函数的源码如下：

publicThreadPoolExecutor(intcorePoolSize,
intmaximumPoolSize,
longkeepAliveTime,
TimeUnitunit,
BlockingQueueworkQueue,
ThreadFactorythreadFactory,
RejectedExecutionHandlerhandler){
if(corePoolSize<0||
maximumPoolSize<=0||
maximumPoolSize
说明：在ThreadPoolExecutor()的构造函数中，进行的是初始化工作。

corePoolSize,maximumPoolSize,unit,keepAliveTime和workQueue这些变量的值是已知的，它们都是通过newFixedThreadPool()传递而来。下面看看threadFactory和handler对象。 
2.1ThreadFactory
线程池中的ThreadFactory是一个线程工厂，线程池创建线程都是通过线程工厂对象(threadFactory)来完成的。

上面所说的threadFactory对象，是通过Executors.defaultThreadFactory()返回的。Executors.java中的defaultThreadFactory()源码如下：
publicstaticThreadFactorydefaultThreadFactory(){
returnnewDefaultThreadFactory();
}
defaultThreadFactory()返回DefaultThreadFactory对象。Executors.java中的DefaultThreadFactory()源码如下：
staticclassDefaultThreadFactoryimplementsThreadFactory{
privatestaticfinalAtomicIntegerpoolNumber=newAtomicInteger(1);
privatefinalThreadGroupgroup;
privatefinalAtomicIntegerthreadNumber=newAtomicInteger(1);
privatefinalStringnamePrefix;

DefaultThreadFactory(){
SecurityManagers=System.getSecurityManager();
group=(s!=null)?s.getThreadGroup():
Thread.currentThread().getThreadGroup();
namePrefix="pool-"+
poolNumber.getAndIncrement()+
"-thread-";
}

//提供创建线程的API。
publicThreadnewThread(Runnabler){
//线程对应的任务是Runnable对象r
Threadt=newThread(group,r,
namePrefix+threadNumber.getAndIncrement(),
0);
//设为“非守护线程”
if(t.isDaemon())
t.setDaemon(false);
//将优先级设为“Thread.NORM_PRIORITY”
if(t.getPriority()!=Thread.NORM_PRIORITY)
t.setPriority(Thread.NORM_PRIORITY);
returnt;
}
}


说明：ThreadFactory的作用就是提供创建线程的功能的线程工厂。

        它是通过newThread()提供创建线程功能的，下面简单说说newThread()。newThread()创建的线程对应的任务是Runnable对象，它创建的线程都是“非守护线程”而且“线程优先级都是Thread.NORM_PRIORITY”。 
2.2RejectedExecutionHandler
handler是ThreadPoolExecutor中拒绝策略的处理句柄。所谓拒绝策略，是指将任务添加到线程池中时，线程池拒绝该任务所采取的相应策略。

线程池默认会采用的是defaultHandler策略，即AbortPolicy策略。在AbortPolicy策略中，线程池拒绝任务时会抛出异常！

defaultHandler的定义如下：
privatestaticfinalRejectedExecutionHandlerdefaultHandler=newAbortPolicy();

AbortPolicy的源码如下：
publicstaticclassAbortPolicyimplementsRejectedExecutionHandler{
publicAbortPolicy(){}

//抛出异常
publicvoidrejectedExecution(Runnabler,ThreadPoolExecutore){
thrownewRejectedExecutionException("Task"+r.toString()+
"rejectedfrom"+
e.toString());
}
}


(二)添加任务到“线程池”
1.execute()
execute()定义在ThreadPoolExecutor.java中，源码如下：
publicvoidexecute(Runnablecommand){
//如果任务为null，则抛出异常。
if(command==null)
thrownewNullPointerException();
//获取ctl对应的int值。该int值保存了"线程池中任务的数量"和"线程池状态"信息
intc=ctl.get();
//当线程池中的任务数量<"核心池大小"时，即线程池中少于corePoolSize个任务。
//则通过addWorker(command,true)新建一个线程，并将任务(command)添加到该线程中；然后，启动该线程从而执行任务。
if(workerCountOf(c)="核心池大小"时，
//而且，"线程池处于允许状态"时，则尝试将任务添加到阻塞队列中。
if(isRunning(c)&&workQueue.offer(command)){
//再次确认“线程池状态”，若线程池异常终止了，则删除任务；然后通过reject()执行相应的拒绝策略的内容。
intrecheck=ctl.get();
if(!isRunning(recheck)&&remove(command))
reject(command);
//否则，如果"线程池中任务数量"为0，则通过addWorker(null,false)尝试新建一个线程，新建线程对应的任务为null。
elseif(workerCountOf(recheck)==0)
addWorker(null,false);
}
//通过addWorker(command,false)新建一个线程，并将任务(command)添加到该线程中；然后，启动该线程从而执行任务。
//如果addWorker(command,false)执行失败，则通过reject()执行相应的拒绝策略的内容。
elseif(!addWorker(command,false))
reject(command);
}

说明：execute()的作用是将任务添加到线程池中执行。它会分为3种情况进行处理：

       情况1--如果"线程池中任务数量"<"核心池大小"时，即线程池中少于corePoolSize个任务；此时就新建一个线程，并将该任务添加到线程中进行执行。

       情况2--如果"线程池中任务数量">="核心池大小"，并且"线程池是允许状态"；此时，则将任务添加到阻塞队列中阻塞等待。在该情况下，会再次确认"线程池的状态"，如果"第2次读到的线程池状态"和"第1次读到的线程池状态"不同，则从阻塞队列中删除该任务。

       情况3--非以上两种情况。在这种情况下，尝试新建一个线程，并将该任务添加到线程中进行执行。如果执行失败，则通过reject()拒绝该任务。
2.addWorker()
addWorker()的源码如下：
privatebooleanaddWorker(RunnablefirstTask,booleancore){
retry:
//更新"线程池状态和计数"标记，即更新ctl。
for(;;){
//获取ctl对应的int值。该int值保存了"线程池中任务的数量"和"线程池状态"信息
intc=ctl.get();
//获取线程池状态。
intrs=runStateOf(c);

//有效性检查
if(rs>=SHUTDOWN&&
!(rs==SHUTDOWN&&
firstTask==null&&
!workQueue.isEmpty()))
returnfalse;

for(;;){
//获取线程池中任务的数量。
intwc=workerCountOf(c);
//如果"线程池中任务的数量"超过限制，则返回false。
if(wc>=CAPACITY||
wc>=(core?corePoolSize:maximumPoolSize))
returnfalse;
//通过CAS函数将c的值+1。操作失败的话，则退出循环。
if(compareAndIncrementWorkerCount(c))
breakretry;
c=ctl.get();//Re-readctl
//检查"线程池状态"，如果与之前的状态不同，则从retry重新开始。
if(runStateOf(c)!=rs)
continueretry;
//elseCASfailedduetoworkerCountchange;retryinnerloop
}
}

booleanworkerStarted=false;
booleanworkerAdded=false;
Workerw=null;
//添加任务到线程池，并启动任务所在的线程。
try{
finalReentrantLockmainLock=this.mainLock;
//新建Worker，并且指定firstTask为Worker的第一个任务。
w=newWorker(firstTask);
//获取Worker对应的线程。
finalThreadt=w.thread;
if(t!=null){
//获取锁
mainLock.lock();
try{
intc=ctl.get();
intrs=runStateOf(c);

//再次确认"线程池状态"
if(rslargestPoolSize)
largestPoolSize=s;
workerAdded=true;
}
}finally{
//释放锁
mainLock.unlock();
}
//如果"成功将任务添加到线程池"中，则启动任务所在的线程。
if(workerAdded){
t.start();
workerStarted=true;
}
}
}finally{
if(!workerStarted)
addWorkerFailed(w);
}
//返回任务是否启动。
returnworkerStarted;
}


说明：

   addWorker(RunnablefirstTask,booleancore)的作用是将任务(firstTask)添加到线程池中，并启动该任务。

   core为true的话，则以corePoolSize为界限，若"线程池中已有任务数量>=corePoolSize"，则返回false；core为false的话，则以maximumPoolSize为界限，若"线程池中已有任务数量>=maximumPoolSize"，则返回false。

   addWorker()会先通过for循环不断尝试更新ctl状态，ctl记录了"线程池中任务数量和线程池状态"。

   更新成功之后，再通过try模块来将任务添加到线程池中，并启动任务所在的线程。
   从addWorker()中，我们能清晰的发现：线程池在添加任务时，会创建任务对应的Worker对象；而一个Workder对象包含一个Thread对象。(01)通过将Worker对象添加到"线程的workers集合"中，从而实现将任务添加到线程池中。(02)通过启动Worker对应的Thread线程，则执行该任务。
 3.submit()
补充说明一点，submit()实际上也是通过调用execute()实现的，源码如下：
publicFuturesubmit(Runnabletask){
if(task==null)thrownewNullPointerException();
RunnableFutureftask=newTaskFor(task,null);
execute(ftask);
returnftask;
}

 (三)关闭“线程池”
shutdown()的源码如下：
publicvoidshutdown(){
finalReentrantLockmainLock=this.mainLock;
//获取锁
mainLock.lock();
try{
//检查终止线程池的“线程”是否有权限。
checkShutdownAccess();
//设置线程池的状态为关闭状态。
advanceRunState(SHUTDOWN);
//中断线程池中空闲的线程。
interruptIdleWorkers();
//钩子函数，在ThreadPoolExecutor中没有任何动作。
onShutdown();//hookforScheduledThreadPoolExecutor
}finally{
//释放锁
mainLock.unlock();
}
//尝试终止线程池
tryTerminate();
}

说明：shutdown()的作用是关闭线程池。
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持毛票票。

Java concurrency线程池之线程池原理(二)_动力节点Java学院整理

热门推荐

随机推荐