第十二章 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();//构建任务1
13         Task2 task2 = new Task2();//构建任务2
14         executor.execute(task1);//执行任务1
15         executor.execute(task2);//执行任务2
16     }
17     
18     /*
19      * 基本任务2
20      */
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      * 基本任务2
32      */
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源码解析

第十二章 ThreadPoolExecutor使用与工作机理

标签：

原文地址：http://www.cnblogs.com/java-zhao/p/5146601.html