你的位置:首页 > Java教程

[Java教程]第十二章 ThreadPoolExecutor使用与工作机理


1、最基础的线程池ThreadPoolExecutor

使用方式:

 1 /** 2  * ThreadPoolExecutor测试类 3  * 注意: 4  * 1、ThreadPoolExecutor是一个线程池 5  * 2、多个任务都可以由该线程池中选出几条线程来执行 6 */ 7 public class ThreadPoolExecutorTest { 8   private static ThreadPoolExecutor executor =  9       new ThreadPoolExecutor(5, 10, 30, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(10));10   11   public void executeTask(){12     Task1 task1 = new Task1();//构建任务113     Task2 task2 = new Task2();//构建任务214     executor.execute(task1);//执行任务115     executor.execute(task2);//执行任务216   }17   18   /*19    * 基本任务220   */21   class Task1 implements Runnable{22     public void run() {23       //具体任务的业务24       for(int i=0;i<1000;i++){25         System.out.println("hello xxx!!!");26       }27     }28   }29   30   /*31    * 基本任务232   */33   class Task2 implements Runnable{34     public void run() {35       //具体任务的业务36       for(int i=0;i<5;i++){37         System.out.println("hello world2!!!");38       }39     }40   }41   42   public static void main(String[] args) {43     ThreadPoolExecutorTest test = new ThreadPoolExecutorTest();44     test.executeTask();45   }46 }

View Code

说明:

在代码中,构建了一个线程池(executor)和两个实现了Runnable接口的任务(task1、task2),并将这两个任务提交到executor中去执行。

线程池的配置:集合下边的工作机理与参数详细说明来说。

当然,上述的执行结果是交叉着的,因为存在线程的切换。

 

2、工作机理

A、当一个新的任务被提交到ThreadPoolExecutor的execute()方法中时,如果当前池中正在运行的线程少于corePoolSize,则会创建一个新的线程来处理该任务;

注意:这是池中正在运行的线程,为什么这样说呢?是因为核心线程是每来一个任务才创建一个线程,这个看第三部分。看完第三部分之后,你就会觉得,其实换个说法:"如果当前池中的线程少于corePoolSize"这样会更准确,因为我们也许会通过下边介绍的方法提前将核心线程创建好,如果假设这时候来了一个任务,而所有的核心线程都处于空闲状态的话,这时候是不会去创建新线程的。

B、如果当前池中的线程大于等于corePoolSize,但是小于maximumPoolSize时,如果队列满了,会创建新的线程来处理任务,如果队列没有满,任务加入到队列中去;

C、如果队列满了,正在运行的线程数已经等于maximumPoolSize时,该任务就会被rejected(回绝)

 

3、参数详细说明

A、corePoolSize与maximumPoolSize

  • 如果corePoolSize==maximumPoolSize,线程池的size就是固定的了(这一块儿类似于堆内存的指定,防止扩张带来的损耗,但要视情况而定);
  • 默认情况下,只有当一个新的任务到达时,才会创建和启动core threads,但是可以通过prestartCoreThread和prestartAllCoreThreads来改变;

B、ThreadFactory

  • 通过使用java.util.concurrent.ThreadFactory可以创建新的线程
  • 如果不额外指定ThreadFactory,则使用默认的Executors#defaultThreadFactory
  • 通过该默认的线程工厂,所有创建的线程都会被加入到同一个ThreadGroup中去,并且这些线程都会有相同的优先级(NORM_PRIORITY),并且都是non-daemon线程

注意:这一块儿有一个后台(daemon)线程的概念,典型的后台线程:垃圾回收线程;这个线程与其他应用线程的不同之处在于:当所有的应用线程都没有后,后台线程也就自动消失了。

C、keepAliveTime

  • 如果pool当前拥有的线程超过了corePoolSize,超出的线程如果在大于keepAliveTime的时间外闲置(idle),这些线程就会被终止;
  • 该机制在pool没有被活跃的使用的时候,可以减少资源浪费;
  • 默认情况下,keep-alive机制仅仅会在线程数超过corePoolSizeThreads时才会被使用;
  • 当然,通过使用ThreadPoolExecutor#allowCoreThreadTimeOut(boolean)也可以将这种keep-alive机制应用在core threads上去(只要keepAliveTime>0即可)

D、Queue

任何一种BlockingQueue都可以被用来传递和存储提交到线程池中的任务,有三种队列策略:

1)SynchronousQueue(默认)

  • 直接将任务移交给线程而不是入队,如果已经没有线程立即来处理提交到pool中的任务时,会创建一个新的线程来处理该任务;
  • 这种策略需要maximumPoolSizes无界来确保新提交的任务不会被rejection;
  • 这种方式的最大缺点:当任务到来的速度大于任务被处理的速度时,线程数会疯长。

2)无界队列LinkedBlockingQueue:

  • 由于队列无界,当运行的线程等于corePoolSize时,新到来的任务会入队而不会创建新的线程来执行(即pool中的线程数永远不会大于corePoolSize);
  • 这种方式的缺点:当任务到来的速度大于任务被处理的速度时,队列长度会疯长。

3)有界队列ArrayBlockingQueue:

  • 这种方式是非常难处理好的一种方式,要考虑好ArrayBlockingQueue的大小和maximumPoolSize的大小;
  • 当ArrayBlockingQueue较大而maximumPoolSize较小时,会降低CPU使用率、减少OS资源、减少上下文切换,但是吞吐量会降低。-->线程较少的特点就是这样;
  • 如果任务频繁的被阻塞(例如,they are I/O bound),就需要更多的线程了;
  • 当ArrayBlockingQueue较小而maximumPoolSize较大时,会使CPU使用繁忙但也会遇到一些不可接受的scheduling,吞吐量也会降低。

说明:这一块儿配置是一个比较麻烦的地方,后边会说。

E、回绝任务

执行回绝的场景:看开头部分的工作机理。

在回绝任务的时候,execute()方法会调用RejectedExecutionHandler#rejectedExecution。有四种handler策略:

1)ThreadPoolExecutor.CallerRunsPolicy:调用execute()的线程自己来处理该任务

2)ThreadPoolExecutor.DiscardPolicy:不能被执行的任务会直接被扔掉

3)ThreadPoolExecutor.DiscardOldestPolicy:如果executor没有被关闭,队列头部的任务将会被丢弃,然后将该任务加到队尾

4)ThreadPoolExecutor.AbortPolicy(默认):回绝任务并抛出异常

F、AOP

ThreadPoolExecutor提供了两个方法在每个任务的执行前后进行调用ThreadPoolExecutor#beforeExecute和ThreadPoolExecutor#afterExecute.

 

4、开头实例套用

实例中构建的线程池参数:

  • corePoolSize==5    
  • maximumPoolSize==10  
  • keepAliveTime==30s 
  • 队列:ArrayBlockingQueue,大小为10
  • 线程工厂:defaultThreadFactory(默认)
  • 回绝策略:AbortPolicy(默认)

套一下工作机理:

1)当并发提交了<=5个任务到executor中时(此时任务数<=corePoolSize),executor会使用5个核心线程去执行这些任务;

2)当这时候马上又来了一个任务,如果此时5个核心线程有空闲线程的话,就是用空闲的线程去处理,如果都在忙,这时候该任务进入队列;

3)之后再来任务,还是像第二步那样去执行,直到任务将队列放满了,这时候,如果再来一个任务,如果5个核心线程有空闲线程,直接去执行该任务(注意:这里使用了非公平模式,如果使用了公平模式的话,应该是先去处理队列头部的任务,然后将该任务加入队尾?这一块儿暂时有疑问,我再看看源代码再回来改),如果5个核心线程都在忙,这时候就创建一个新的线程来执行该任务;

4)如果通过上边的流程,最后5个线程都在忙,并且队列满了,并且pool中的线程数已经是10个了(池中的线程总数==maximumPoolSize了),这时候就要执行回绝策略了,在这里,使用了默认的AbortPolicy,即直接放弃该任务并抛出异常。

在代码的执行过程中,如果发现后来创建的5个线程有超过30秒都没被调用过的,该线程就被回收掉了。

 

疑问:(这个疑问我会在看完ThreadPoolExecutor的相关源码后进行回答)

当队列满了之后,这时候来了一个任务,恰好5个核心线程有一个空闲了,那么下面两种情况哪一个正确:

1)这个空闲的核心线程直接执行刚刚到来的任务

2)这个空闲的核心线程直接执行队列头部的任务,而将刚刚到来的任务放入队尾

 

最后,这里列出上边提到的两种队列的源码解析地址:

ArrayBlockingQueue:第八章 ArrayBlockingQueue源码解析  

LinkedBlockingQueue:第九章 LinkedBlockingQueue源码解析