频道栏目
IT货架 > > 正文
Executor-ThreadPoolExecutor实现
网友分享于:Jan 1, 1970 8:00:00 AM    来源: IT货架   

标签:executor   threadpoolexecutor   

 1、ThreadPoolExecutor的主要作用

在Oracle中对ThreadPoolExecutor的作用进行了说明:1、在大量的异步任务到达的情况下,使用线程池能够提升性能;2、提供一种资源管理和调度的方法。

一般通过Executors的工厂方法来生成一个线程池对象,Executors提供了多种方法来构造不同的线程池:1、带有缓存性质的线程池  Executors.newCachedThreadPool(),线程池的大小不固定,并且会随着使用情况自动调整线程池的大小;2、固定大小的线程池 Executors.newFixedThreadPool(int);3、单一线程池,只有一个后台线程, Executors.newSingleThreadExecutor()

2、线程池控制属性

在使用ThreadPoolExecutor可以指定一些参数来对线程池进行控制。

    当前线程数 poolSize  

线程池中当前线程的个数

     核心线程数  corePoolSize

线程池中核心线程的个数,也就是能够长期存活的线程个数

     最大线程数 maximumPoolSize

corePoolSize和maximumPoolSize二者之间共同合作来管理线程池中线程的数量,同时能够控制线程池的行为。当一个任务被提交到线程池中,根据当前线程池中线程数目的不同,线程池的处理也不一样。当前线程数用poolSize表示

   最大线程数 largestPoolSize

largestPoolSize并不参与线程池的控制,它用来记录,线程池中曾经同时运行的最大线程数目。


     等待时间 keepAliveTime

等待时间是用来控制超过corePoolSize部分的线程的空闲时间。当线程池中的线程数超过了corePoolSize时,超出部分的线程在空闲时间达到了keepAliveTime指定的时候之后,就会被终止,这样当对线程池使用不是很多的时候,可以节省机器资源。

3、线程池的状态

RUNNING 可以接受新任务
SHUTDOWN 不接受新任务,正在排队的已提交任务会去执行
STOP 不接受新任务,不执行已提交的正在排队的任务,同时中断所有正在执行的任务
TERMINATED  除了所有任务已经终止外,其它表象和STOP相同。
     

4、线程池的调度

当一个任务进入线程时, 根据线程池中当前线程的数据和配置的corePoolSize以及maximumPoolSize的值,线程池执行的动作也不一样。

a、poolSize < corePoolSize
    当线程池中的数目小于指定的核心线程数时,线程池会为新提交的任务创建一个线程来执行任务。

b、poolSize>=corePoolSize

     如果可以放入等待队列中,则放在等待队列中,等待其它线程执行完任务之后来执行此任务
     如果无法放入等待队列中,则检查poolSize是否小于maximum,小于的话则新建一个线程来执行任务。如果已经达到线程数上限了,就会提示用户拒绝任务提交。

bubuko.com,布布扣

bubuko.com,布布扣

5、实现
    
首先来看ThreadPoolSize中定义的几个属性

private  final  BlockingQueue<Runnable> workQueue;
workQueue是已经提交到线程池,但还未来得及执行的任务。

private  final  HashSet<Worker>  workers  =  new  HashSet<Worker>();
workers中存放线程池中当前正在工作的线程信息。ThreadPoolExecutor使用Worker来对任务进行了包装。

private  volatile  long    keepAliveTime  ;
private  volatile  int     corePoolSize  ;
private  volatile  int     maximumPoolSize  ;
private  volatile  int     poolSize  ;
private  int  largestPoolSize ;


还有一些属性,对于理解整个线程池的运作原理不那么重要,这里就没有列出。

任务进入线程池

在《Executor实现----AbstractExecutorService实现分析》中分析过,在submit提交任务中,最终是通过线程池的execute方法来实现Executors框架对任务的执行。
下面来看ThreadPoolExecutor的具体实现
public  void  execute (Runnable command) {
         if  (command ==  null )
             throw  new  NullPointerException();
         if  ( poolSize  >=  corePoolSize  || !addIfUnderCorePoolSize(command)) {
             if  ( runState  ==  RUNNING  &&  workQueue .offer(command)) {
                 if  ( runState  !=  RUNNING  ||  poolSize  == 0)
                    ensureQueuedTaskHandled(command);
            }
             else  if  (!addIfUnderMaximumPoolSize(command))
                reject(command);  // is shutdown or saturated
        }
    }
通过 if  ( poolSize  >=  corePoolSize  || !addIfUnderCorePoolSize(command)) 来判断当前是否创建一个线程来执行任务。

如果线程池中线程数还没有达到核心线程数,就直接调用addIfUnderCorePoolSize来创建线程执行任务。

如果线程池中线程的个数已经达到了核心线程数,就使用workQueue.offer(command)把任务放入等待队列中。

如果workQueue.offer操作返回false,表示等待队列已满,无法放入更多任务,调用  addIfUnderMaximumPoolSize来检查当前是否已经达到最大线程数,是否还可以创建新的线程来执行任务。

addIfUnderCorePoolSize和 addIfUnderMaximumPoolSize的代码如下

private  boolean  addIfUnderCorePoolSize (Runnable firstTask) {
        Thread t =  null ;
         final  ReentrantLock mainLock =  this . mainLock ;
        mainLock.lock();
         try  {
             if  ( poolSize  < corePoolSize &&  runState  ==  RUNNING )
                t = addThread(firstTask);
        }  finally  {
            mainLock.unlock();
        }
         return  t !=  null  ;
    }

private  boolean  addIfUnderMaximumPoolSize(Runnable firstTask) {
        Thread t =  null ;
         final  ReentrantLock mainLock =  this . mainLock ;
        mainLock.lock();
         try  {
             if  ( poolSize  <  maximumPoolSize  &&  runState  ==  RUNNING )
                t = addThread(firstTask);
        }  finally  {
            mainLock.unlock();
        }
         return  t !=  null  ;
    }

检查poolSize是否满足创建线程的条件,然后调 用addThread来创建线程去执行任务。下面来看下addThread方法的具体实现
private  Thread addThread(Runnable firstTask) {
        Worker w =  new  Worker(firstTask);
        Thread t =  threadFactory .newThread(w);
         boolean  workerStarted =  false ;
         if  (t !=  null  ) {
             if  (t.isAlive())  // precheck that t is startable
                 throw  new  IllegalThreadStateException();
            w.  thread  = t;
             workers .add(w);
             int  nt = ++ poolSize  ;
             if  (nt >  largestPoolSize  )
                 largestPoolSize  = nt;
             try  {
                t.start();
                workerStarted =  true ;
            }
             finally  {
                 if  (!workerStarted)
                     workers .remove(w);
            }
        }
         return  t;
    }
首先threadFactory.newThread创建一个新线程,然后把任务添加workers中。把任务封装成一个Worker对象,然后把这个Worker对象封装成一个Thread对象。


然后启动这个线程。注意这里对largestPoolSize的设置,从这里可以看出,只要当前线程池中线程数大于largestPoolSize,就更新largestPoolSize字段,也就是用largestPoolSize来记录线程池中历史线程最大数。

从addThread中可以看出,任务的执行最终是通过Wokrer来实现的。那一个问题是,当前这个任务执行完成后,线程是销毁,还是继续执行?如果执行的话,又如何取任务?在线程池中线程数已经超过corePoolSize的情况下,线程池又是如何控制线程池中线程的数目的?

这些都需要通过Worker来了解, 下面来看Worker类的具体实现。
private  final  class  Worker  implements  Runnable

Worker实现了Runnable接口,所以才能被封装到线程中(  Thread t =  threadFactory  .newThread(w);就是做的这样的封装 ),在线程启动的时候,执行run方法。

private  Runnable  firstTask ;
Thread  thread ;
volatile  boolean  hasRun  =  false ;

这是Worker的三个属性(还有其它几个属性没有列出来)。firstTask初始化Worker对象时执行的任务,也就是worker启动时要执行的任务。thread是worker任务执行所在的线程,每个worker都属于一个线程,在这个线程中执行。hasRun是当前任务的执行状态,表示任务是不是已经执行完成

开门见山,直接看Worker的run方法
  public  void  run() {
             try  {
                 hasRun  =  true ;
                Runnable task =  firstTask ;
                 firstTask  =  null  ;
                 while  (task !=  null  || (task = getTask()) !=  null ) {
                    runTask(task);
                    task =  null ;
                }
            }  finally  {
                workerDone(  this );
            }
        }
    }
注意这里的while循环,取任务--执行任务--取任务---执行任务  正常情况下,只要能正常获取到任务,这个线程就会一直执行下去。
所以,当线程池中提交的任务源源不断时,线程池中的线程数就不会减少。那么当线程池中提交的任务减少,while循环的条件不满足getTask返回null,或者抛出异常,就会到workerDone,来实现线程池中多余线程的销毁,节省资源。

OK,这里就有个问题,如果这样的话,要如何实现线程池中corePoolSize线程数的长期存活呢?如果实现超出corePoolSize的部分才会在空闲一段时间后销毁?答案在getTask中

 Runnable getTask() {
         for  (;;) {
             try  {
                 int  state =  runState  ;
                 if  (state >  SHUTDOWN )
                     return  null  ;
                Runnable r;
                 if  (state ==  SHUTDOWN   // Help drain queue
                    r =  workQueue .poll();
                 else  if  ( poolSize  >  corePoolSize  ||  allowCoreThreadTimeOut  )
                    r =  workQueue .poll( keepAliveTime  , TimeUnit. NANOSECONDS );
                 else
                    r =  workQueue .take();
                 if  (r !=  null  )
                     return  r;
                 if  (workerCanExit()) {
                     if  ( runState  >=  SHUTDOWN )  // Wake up others
                        interruptIdleWorkers();
                     return  null  ;
                }
                 // Else retry
            }  catch  (InterruptedException ie) {
                 // On interruption, re-check runState
            }
        }
    }

注意当poolSize>corePoolSize时,从任务队列中获取任务时,是通过poll,并指定了超时时间(keepAliveTime在这里起效),这样当线程池中任务较少时,这里会返回null,workerCanExit返回true,所以整个getTask就返回null,这样就使得while循环条件不为true,从而执行workerDone,结束这个超出核心线程数的线程。

而在poolSize<=corePoolSize时,take使得线程阻塞。直到有新任务到来为止。

通过这两种不同的情况,来实现根据当前线程池中任务的多少去进行超出corePoolSize的线程的动态调整。
下面是一个流程示意图
bubuko.com,布布扣
bubuko.com,布布扣

Executor-ThreadPoolExecutor实现

标签:executor   threadpoolexecutor   

广告服务联系QQ:1134687142 | 网站地图

版权所有: IT货架- 内容来自互联网,仅供用于技术学习,请遵循相关法律法规. 京ICP备11030978号-1