深入理解Java之线程池(中)

接上文

下面我们看一下Worker类的实现:

private final class Worker implements Runnable {

private final ReentrantLock runLock = new ReentrantLock();

private Runnable firstTask;

volatile long completedTasks;

Thread thread;

Worker(Runnable firstTask) {

this.firstTask = firstTask;

}

boolean isActive() {

return runLock.isLocked();

}

void interruptIfIdle() {

final ReentrantLock runLock = this.runLock;

if (runLock.tryLock()) {

try {

if (thread != Thread.currentThread())

thread.interrupt();

} finally {

runLock.unlock();

}

}

}

void interruptNow() {

thread.interrupt();

}

 

private void runTask(Runnable task) {

final ReentrantLock runLock = this.runLock;

runLock.lock();

try {

if (runState < STOP &&

Thread.interrupted() &&

runState >= STOP)

boolean ran = false;

beforeExecute(thread, task);   //beforeExecute方法是ThreadPoolExecutor类的一个方法,没有具体实现,用户可以根据

//自己需要重载这个方法和后面的afterExecute方法来进行一些统计信息,比如某个任务的执行时间等

try {

task.run();

ran = true;

afterExecute(task, null);

++completedTasks;

} catch (RuntimeException ex) {

if (!ran)

afterExecute(task, ex);

throw ex;

}

} finally {

runLock.unlock();

}

}

 

public void run() {

try {

Runnable task = firstTask;

firstTask = null;

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

runTask(task);

task = null;

}

} finally {

workerDone(this);   //当任务队列中没有任务时,进行清理工作

}

}

}

它实际上实现了Runnable接口,因此上面的Thread t = threadFactory.newThread(w);效果跟下面这句的效果基本一样:

Thread t = new Thread(w);

相当于传进去了一个Runnable任务,在线程t中执行这个Runnable。

既然Worker实现了Runnable接口,那么自然最核心的方法便是run()方法了:

public void run() {

try {

Runnable task = firstTask;

firstTask = null;

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

runTask(task);

task = null;

}

} finally {

workerDone(this);

}

}

从run方法的实现可以看出,它首先执行的是通过构造器传进来的任务firstTask,在调用runTask()执行完firstTask之后,在while循环里面不断通过getTask()去取新的任务来执行,那么去哪里取呢?自然是从任务缓存队列里面去取,getTask是ThreadPoolExecutor类中的方法,并不是Worker类中的方法,下面是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) //如果线程数大于核心池大小或者允许为核心池线程设置空闲时间,

//则通过poll取任务,若等待一定的时间取不到任务,则返回null

r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS);

else

r = workQueue.take();

if (r != null)

return r;

if (workerCanExit()) {    //如果没取到任务,即r为null,则判断当前的worker是否可以退出

if (runState >= SHUTDOWN) // Wake up others

interruptIdleWorkers();   //中断处于空闲状态的worker

return null;

}

// Else retry

} catch (InterruptedException ie) {

// On interruption, re-check runState

}

}

}

在getTask中,先判断当前线程池状态,如果runState大于SHUTDOWN(即为STOP或者TERMINATED),则直接返回null。

如果runState为SHUTDOWN或者RUNNING,则从任务缓存队列取任务。

如果当前线程池的线程数大于核心池大小corePoolSize或者允许为核心池中的线程设置空闲存活时间,则调用poll(time,timeUnit)来取任务,这个方法会等待一定的时间,如果取不到任务就返回null。

然后判断取到的任务r是否为null,为null则通过调用workerCanExit()方法来判断当前worker是否可以退出,我们看一下workerCanExit()的实现:

private boolean workerCanExit() {

final ReentrantLock mainLock = this.mainLock;

mainLock.lock();

boolean canExit;

//如果runState大于等于STOP,或者任务缓存队列为空了

//或者  允许为核心池线程设置空闲存活时间并且线程池中的线程数目大于1

try {

canExit = runState >= STOP ||

workQueue.isEmpty() ||

(allowCoreThreadTimeOut &&

poolSize > Math.max(1, corePoolSize));

} finally {

mainLock.unlock();

}

return canExit;

}

也就是说如果线程池处于STOP状态、或者任务队列已为空或者允许为核心池线程设置空闲存活时间并且线程数大于1时,允许worker退出。如果允许worker退出,则调用interruptIdleWorkers()中断处于空闲状态的worker,我们看一下interruptIdleWorkers()的实现:

void interruptIdleWorkers() {

final ReentrantLock mainLock = this.mainLock;

mainLock.lock();

try {

for (Worker w : workers)  //实际上调用的是worker的interruptIfIdle()方法

w.interruptIfIdle();

} finally {

mainLock.unlock();

}

}

从实现可以看出,它实际上调用的是worker的interruptIfIdle()方法,在worker的interruptIfIdle()方法中:

void interruptIfIdle() {

final ReentrantLock runLock = this.runLock;

if (runLock.tryLock()) {    //注意这里,是调用tryLock()来获取锁的,因为如果当前worker正在执行任务,锁已经被获取了,是无法获取到锁的

//如果成功获取了锁,说明当前worker处于空闲状态

try {

if (thread != Thread.currentThread())

thread.interrupt();

} finally {

runLock.unlock();

}

}

}

这里有一个非常巧妙的设计方式,假如我们来设计线程池,可能会有一个任务分派线程,当发现有线程空闲时,就从任务缓存队列中取一个任务交给空闲线程执行。但是在这里,并没有采用这样的方式,因为这样会要额外地对任务分派线程进行管理,无形地会增加难度和复杂度,这里直接让执行完任务的线程去任务缓存队列里面取任务来执行。

我们再看addIfUnderMaximumPoolSize方法的实现,这个方法的实现思想和addIfUnderCorePoolSize方法的实现思想非常相似,唯一的区别在于addIfUnderMaximumPoolSize方法是在线程池中的线程数达到了核心池大小并且往任务队列中添加任务失败的情况下执行的:

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();

}

if (t == null)

return false;

t.start();

return true;

}

看到没有,其实它和addIfUnderCorePoolSize方法的实现基本一模一样,只是if语句判断条件中的poolSize < maximumPoolSize不同而已。

到这里,大部分朋友应该对任务提交给线程池之后到被执行的整个过程有了一个基本的了解,下面总结一下:

1)首先,要清楚corePoolSize和maximumPoolSize的含义;

2)其次,要知道Worker是用来起到什么作用的;

3)要知道任务提交给线程池之后的处理策略,这里总结一下主要有4点:

  • 如果当前线程池中的线程数目小于corePoolSize,则每来一个任务,就会创建一个线程去执行这个任务;
  • 如果当前线程池中的线程数目>=corePoolSize,则每来一个任务,会尝试将其添加到任务缓存队列当中,若添加成功,则该任务会等待空闲线程将其取出去执行;若添加失败(一般来说是任务缓存队列已满),则会尝试创建新的线程去执行这个任务;
  • 如果当前线程池中的线程数目达到maximumPoolSize,则会采取任务拒绝策略进行处理;
  • 如果线程池中的线程数量大于 corePoolSize时,如果某线程空闲时间超过keepAliveTime,线程将被终止,直至线程池中的线程数目不大于corePoolSize;如果允许为核心池中的线程设置存活时间,那么核心池中的线程空闲时间超过keepAliveTime,线程也会被终止。

3.线程池中的线程初始化

默认情况下,创建线程池之后,线程池中是没有线程的,需要提交任务之后才会创建线程。

在实际中如果需要线程池创建之后立即创建线程,可以通过以下两个方法办到:

  • prestartCoreThread():初始化一个核心线程;
  • prestartAllCoreThreads():初始化所有核心线程

下面是这2个方法的实现:

public boolean prestartCoreThread() {

return addIfUnderCorePoolSize(null); //注意传进去的参数是null

}

 

public int prestartAllCoreThreads() {

int n = 0;

while (addIfUnderCorePoolSize(null))//注意传进去的参数是null

++n;

return n;

}

注意上面传进去的参数是null,根据第2小节的分析可知如果传进去的参数为null,则最后执行线程会阻塞在getTask方法中的

r = workQueue.take();

即等待任务队列中有任务。

4.任务缓存队列及排队策略

在前面我们多次提到了任务缓存队列,即workQueue,它用来存放等待执行的任务。

workQueue的类型为BlockingQueue<Runnable>,通常可以取下面三种类型:

1)ArrayBlockingQueue:基于数组的先进先出队列,此队列创建时必须指定大小;

2)LinkedBlockingQueue:基于链表的先进先出队列,如果创建时没有指定此队列大小,则默认为Integer.MAX_VALUE;

3)synchronousQueue:这个队列比较特殊,它不会保存提交的任务,而是将直接新建一个线程来执行新来的任务。

5.任务拒绝策略

当线程池的任务缓存队列已满并且线程池中的线程数目达到maximumPoolSize,如果还有任务到来就会采取任务拒绝策略,通常有以下四种策略:

ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。

ThreadPoolExecutor.DiscardPolicy:也是丢弃任务,但是不抛出异常。

ThreadPoolExecutor.DiscardOldestPolicy:丢弃队列最前面的任务,然后重新尝试执行任务(重复此过程)

ThreadPoolExecutor.CallerRunsPolicy:由调用线程处理该任务

6.线程池的关闭

ThreadPoolExecutor提供了两个方法,用于线程池的关闭,分别是shutdown()和shutdownNow(),其中:

  • shutdown():不会立即终止线程池,而是要等所有任务缓存队列中的任务都执行完后才终止,但再也不会接受新的任务
  • shutdownNow():立即终止线程池,并尝试打断正在执行的任务,并且清空任务缓存队列,返回尚未执行的任务

7.线程池容量的动态调整

ThreadPoolExecutor提供了动态调整线程池容量大小的方法:setCorePoolSize()和setMaximumPoolSize(),

  • setCorePoolSize:设置核心池大小
  • setMaximumPoolSize:设置线程池最大能创建的线程数目大小

当上述参数从小变大时,ThreadPoolExecutor进行线程赋值,还可能立即创建新的线程来执行任务。

三.使用示例

前面我们讨论了关于线程池的实现原理,这一节我们来看一下它的具体使用:

public class Test {

public static void main(String[] args) {

ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 10, 200, TimeUnit.MILLISECONDS,

new ArrayBlockingQueue<Runnable>(5));

 

for(int i=0;i<15;i++){

MyTask myTask = new MyTask(i);

executor.execute(myTask);

System.out.println(“线程池中线程数目:”+executor.getPoolSize()+”,队列中等待执行的任务数目:”+

executor.getQueue().size()+”,已执行玩别的任务数目:”+executor.getCompletedTaskCount());

}

executor.shutdown();

}

}

 

class MyTask implements Runnable {

private int taskNum;

 

public MyTask(int num) {

this.taskNum = num;

}

 

@Override

public void run() {

System.out.println(“正在执行task “+taskNum);

try {

Thread.currentThread().sleep(4000);

} catch (InterruptedException e) {

e.printStackTrace();

}

System.out.println(“task “+taskNum+”执行完毕”);

}

}

执行结果:

正在执行task 0

线程池中线程数目:1,队列中等待执行的任务数目:0,已执行玩别的任务数目:0

线程池中线程数目:2,队列中等待执行的任务数目:0,已执行玩别的任务数目:0

正在执行task 1

线程池中线程数目:3,队列中等待执行的任务数目:0,已执行玩别的任务数目:0

正在执行task 2

线程池中线程数目:4,队列中等待执行的任务数目:0,已执行玩别的任务数目:0

正在执行task 3

线程池中线程数目:5,队列中等待执行的任务数目:0,已执行玩别的任务数目:0

正在执行task 4

线程池中线程数目:5,队列中等待执行的任务数目:1,已执行玩别的任务数目:0

线程池中线程数目:5,队列中等待执行的任务数目:2,已执行玩别的任务数目:0

线程池中线程数目:5,队列中等待执行的任务数目:3,已执行玩别的任务数目:0

线程池中线程数目:5,队列中等待执行的任务数目:4,已执行玩别的任务数目:0

线程池中线程数目:5,队列中等待执行的任务数目:5,已执行玩别的任务数目:0

线程池中线程数目:6,队列中等待执行的任务数目:5,已执行玩别的任务数目:0

正在执行task 10

线程池中线程数目:7,队列中等待执行的任务数目:5,已执行玩别的任务数目:0

正在执行task 11

线程池中线程数目:8,队列中等待执行的任务数目:5,已执行玩别的任务数目:0

正在执行task 12

线程池中线程数目:9,队列中等待执行的任务数目:5,已执行玩别的任务数目:0

正在执行task 13

线程池中线程数目:10,队列中等待执行的任务数目:5,已执行玩别的任务数目:0

正在执行task 14

task 3执行完毕

task 0执行完毕

task 2执行完毕

task 1执行完毕

正在执行task 8

正在执行task 7

正在执行task 6

正在执行task 5

task 4执行完毕

task 10执行完毕

task 11执行完毕

task 13执行完毕

task 12执行完毕

正在执行task 9

task 14执行完毕

task 8执行完毕

task 5执行完毕

task 7执行完毕

task 6执行完毕

task 9执行完毕

从执行结果可以看出,当线程池中线程的数目大于5时,便将任务放入任务缓存队列里面,当任务缓存队列满了之后,便创建新的线程。如果上面程序中,将for循环中改成执行20个任务,就会抛出任务拒绝异常了。

不过在java doc中,并不提倡我们直接使用ThreadPoolExecutor,而是使用Executors类中提供的几个静态方法来创建线程池:

Executors.newCachedThreadPool();        //创建一个缓冲池,缓冲池容量大小为Integer.MAX_VALUE

Executors.newSingleThreadExecutor();   //创建容量为1的缓冲池

Executors.newFixedThreadPool(int);    //创建固定容量大小的缓冲池

接下文

Leave a Reply

Fill in your details below or click an icon to log in:

WordPress.com Logo

You are commenting using your WordPress.com account. Log Out /  Change )

Google photo

You are commenting using your Google account. Log Out /  Change )

Twitter picture

You are commenting using your Twitter account. Log Out /  Change )

Facebook photo

You are commenting using your Facebook account. Log Out /  Change )

Connecting to %s