Java并发之——线程池 / 憋错料

一. 线程池介绍

1.1 简介

　　线程池是一种多线程处理形式，处理过程中将任务添加到队列，然后在创建线程后自动启动这些任务。线程池的基本思想还是一种对象池的思想，开辟一块内存空间，里面存放了众多(未死亡)的线程，池中线程执行调度由池管理器来处理。当有线程任务时，从池中取一个，执行完成后线程对象归池，这样可以避免反复创建线程对象所带来的性能开销，节省了系统的资源。

　　多线程技术主要解决处理器单元内多个线程执行的问题，它可以显著减少处理器单元的闲置时间，增加处理器单元的吞吐能力。
　　假设一个服务器完成一项任务所需时间为：T1 创建线程时间，T2 在线程中执行任务的时间，T3 销毁线程时间。如果：T1 + T3 远大于 T2，则可以采用线程池，以提高服务器性能。

　　线程池技术正是关注如何缩短或调整T1,T3时间的技术，从而提高服务器程序性能的。它把T1，T3分别安排在服务器程序的启动和结束的时间段或者一些空闲的时间段，这样在服务器程序处理客户请求时，不会有T1，T3的开销了。线程池不仅调整T1,T3产生的时间段，而且它还显著减少了创建线程的数目，比如：
　　假设一个服务器一天要处理50000个请求，并且每个请求需要一个单独的线程完成。在线程池中，线程数一般是固定的，所以产生线程总数不会超过线程池中线程的数目，而如果服务器不利用线程池来处理这些请求则线程总数为50000。一般线程池大小是远小于50000。所以利用线程池的服务器程序不会为了创建50000而在处理请求时浪费时间，从而提高效率。

1.2 线程池优点

第一：降低资源消耗。通过重复利用已创建的线程降低线程创建和销毁造成的消耗。

第二：提高响应速度。当任务到达时，任务可以不需要等到线程创建就能立即执行。

第三：提高线程的可管理性。线程是稀缺资源，如果无限制的创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一的分配，调优和监控。

二. 线程池的具体类

　　java中常用的线程池类主要有Executors类和ThreadPoolExecutor类。

2.1 Executors类

　　Executors类可以用于方便的创建线程池。它为Executor，ExecutorService，ScheduledExecutorService，ThreadFactory和Callable类提供了一些工具方法。Executors。在java doc中，并不提倡我们直接使用ThreadPoolExecutor，而是使用Executors类中提供的四个静态方法来创建线程池：

2.1.1 newCachedThreadPool

　　创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。

1 public static ExecutorService newCachedThreadPool() {
2     return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
3                                   60L, TimeUnit.SECONDS,
4                                   new SynchronousQueue<Runnable>());
5 }

　　newCachedThreadPool将corePoolSize设置为0，将maximumPoolSize设置为Integer.MAX_VALUE，使用的SynchronousQueue，也就是说来了任务就创建线程运行，当线程空闲超过60秒，就销毁线程。

2.1.2 newFixedThreadPool

创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。

1 public static ExecutorService newFixedThreadPool(int nThreads) {
2     return new ThreadPoolExecutor(nThreads, nThreads,
3                                   0L, TimeUnit.MILLISECONDS,
4                                   new LinkedBlockingQueue<Runnable>());
5 }

　　newFixedThreadPool创建的线程池corePoolSize和maximumPoolSize值是相等的，它使用的LinkedBlockingQueue。

2.1.3 newScheduledThreadPool

创建一个定长线程池，支持定时及周期性任务执行。

1 public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
2     return new ScheduledThreadPoolExecutor(corePoolSize);
3 }

2.1.4 newSingleThreadExecutor

创建一个单线程化的线程池（容量为1的缓冲池），它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。

1 public static ExecutorService newSingleThreadExecutor() {
2     return new FinalizableDelegatedExecutorService
3         (new ThreadPoolExecutor(1, 1,
4                                 0L, TimeUnit.MILLISECONDS,
5                                 new LinkedBlockingQueue<Runnable>()));
6 }

　　newSingleThreadExecutor将corePoolSize和maximumPoolSize都设置为1，也使用的LinkedBlockingQueue。

　　如果Executors提供的几个静态方法能满足要求，就尽量使用它提供的三个方法，因为自己去手动配置ThreadPoolExecutor的参数有点麻烦，要根据实际任务的类型和数量来进行配置。另外，如果ThreadPoolExecutor达不到要求，可以自己继承ThreadPoolExecutor类进行重写。

2.2 ThreadPoolExecutor类

　　java.uitl.concurrent.ThreadPoolExecutor类是线程池中最核心的一个类，因此如果要透彻地了解Java中的线程池，必须先了解这个类。

2.2.1 ThreadPoolExecutor类图关系：

　　Executor是一个顶层接口，在它里面只声明了一个方法execute(Runnable)，返回值为void，参数为Runnable类型，该方法用于接收执行用户提交任务。
　　ExecutorService 接口继承了Executor接口，定义了线程池终止和创建及提交 futureTask 任务支持的方法。并声明了一些方法：submit、invokeAll、invokeAny以及shutDown等。
　　AbstractExecutorService 是抽象类，它实现了ExecutorService接口及其中的的所有方法。主要实现了 ExecutorService 和 futureTask 相关的一些任务创建和提交的方法。
　　ThreadPoolExecutor 继承了类AbstractExecutorService，它是最核心的一个类，是线程池的内部实现。线程池的功能都在这里实现了，平时用的最多的基本就是这个。其源码很精练，远没当时想象的多。
　　ScheduledThreadPoolExecutor 在 ThreadPoolExecutor 的基础上提供了支持定时调度的功能。线程任务可以在一定延时时间后才被触发执行。

2.2.2 ThreadPoolExecutor的构造方法（4个）

 1 public class ThreadPoolExecutor extends AbstractExecutorService {
 2     .....
 3     public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
 4             BlockingQueue<Runnable> workQueue);
 5
 6     public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
 7             BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory);
 8
 9     public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
10             BlockingQueue<Runnable> workQueue,RejectedExecutionHandler handler);
11
12     public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
13         BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler);
14     ...
15 }

　　ThreadPoolExecutor继承了AbstractExecutorService类，并提供了四个构造器，事实上，通过观察每个构造器的源码具体实现，发现前面三个构造器都是调用的第四个构造器进行的初始化工作。

　　下面解释下一下构造器中各个参数的含义：

corePoolSize：核心池的大小，这个参数跟后面讲述的线程池的实现原理有非常大的关系。在创建了线程池后，默认情况下，线程池中并没有任何线程，而是等待有任务到来才创建线程去执行任务，除非调用了prestartAllCoreThreads()或者prestartCoreThread()方法，从这2个方法的名字就可以看出，是预创建线程的意思，即在没有任务到来之前就创建corePoolSize个线程或者一个线程。默认情况下，在创建了线程池后，线程池中的线程数为0，当有任务来之后，就会创建一个线程去执行任务，当线程池中的线程数目达到corePoolSize后，就会把到达的任务放到缓存队列当中。
maximumPoolSize：线程池最大线程数，这个参数也是一个非常重要的参数，它表示在线程池中最多能创建多少个线程。
keepAliveTime：表示线程没有任务执行时最多保持多久时间会终止。默认情况下，只有当线程池中的线程数大于corePoolSize时，keepAliveTime才会起作用，直到线程池中的线程数不大于corePoolSize，即当线程池中的线程数大于corePoolSize时，如果一个线程空闲的时间达到keepAliveTime，则会终止，直到线程池中的线程数不超过corePoolSize。但是如果调用了allowCoreThreadTimeOut(boolean)方法，在线程池中的线程数不大于corePoolSize时，keepAliveTime参数也会起作用，直到线程池中的线程数为0。

unit：参数keepAliveTime的时间单位，有7种取值，在TimeUnit类中有7种静态属性：

1 TimeUnit.DAYS;               //天
2 TimeUnit.HOURS;             //小时
3 TimeUnit.MINUTES;           //分钟
4 TimeUnit.SECONDS;           //秒
5 TimeUnit.MILLISECONDS;      //毫秒
6 TimeUnit.MICROSECONDS;      //微妙
7 TimeUnit.NANOSECONDS;       //纳秒

workQueue：一个阻塞队列，用来存储等待执行的任务，这个参数的选择也很重要，会对线程池的运行过程（线程池的排队策略）产生重大影响，通常可以取下面三种类型ArrayBlockingQueue（基于数组的先进先出队列，此队列创建时必须指定大小）、LinkedBlockingQueue（基于链表的先进先出队列，如果创建时没有指定此队列大小，则默认为Integer.MAX_VALUE）、SynchronousQueue（这个队列比较特殊，它不会保存提交的任务，而是将直接新建一个线程来执行新来的任务）。一般使用LinkedBlockingQueue和Synchronous。
threadFactory：线程工厂，主要用来创建线程；
handler：表示当拒绝处理任务时的策略，有以下四种取值：

ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。
ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常。
ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程）
ThreadPoolExecutor.CallerRunsPolicy：由调用线程处理该任务

2.2.3 ThreadPoolExecutor类中重要的方法

execute()        //execute()方法实际上是Executor中声明的方法，
                 //在ThreadPoolExecutor进行了具体的实现，这个方法是ThreadPoolExecutor的核心方法，
                 //通过这个方法可以向线程池提交一个任务，交由线程池去执行。
submit()         //submit()方法是在ExecutorService中声明的方法，
                 //在AbstractExecutorService就已经有了具体的实现，在ThreadPoolExecutor中并没有对其进行重写，
                 //这个方法也是用来向线程池提交任务的，但是它和execute()方法不同，它能够返回任务执行的结果，
                 //去看submit()方法的实现，会发现它实际上还是调用的execute()方法，只不过它利用了Future来获取任务执行结果。
shutdown()     　//用来关闭线程池
shutdownNow()    //用来关闭线程池

　　除了这几个比较重要的方法之外，ThreadPoolExecutor还有很多其他的方法，getQueue() 、getPoolSize() 、getActiveCount()、getCompletedTaskCount()等。

三. 线程池实现原理（ThreadPoolExecutor）

3.1 ThreadPoolExecutor内部的几个重要属性

3.1.1 线程池状态

　　在ThreadPoolExecutor中定义了一个volatile变量，另外定义了几个static final变量表示线程池的各个状态：

 1 volatile int runState;
 2
 3 static final int RUNNING    = 0;    //当创建线程池后，初始时，线程池处于RUNNING状态。
 4
 5 static final int SHUTDOWN   = 1;    //如果调用了shutdown()方法，则线程池处于SHUTDOWN状态，
 6                                     //此时线程池不能够接受新的任务，它会等待所有任务执行完毕。
 7 static final int STOP       = 2;    //如果调用了shutdownNow()方法，则线程池处于STOP状态，
 8                                     //此时线程池不能接受新的任务，并且会去尝试终止正在执行的任务。
 9 static final int TERMINATED = 3;    //当线程池处于SHUTDOWN或STOP状态，并且所有工作线程已经销毁，
10                                     //任务缓存队列已经清空或执行结束后，线程池被设置为TERMINATED状态。

　　runState表示当前线程池的状态，它是一个volatile变量用来保证线程之间的可见性。以上几个static final变量表示runState可能的几个取值。

3.1.2 等待任务队列和工作集

1 private final BlockingQueue<Runnable> workQueue; //等待被执行的Runnable任务
2 private final HashSet<Worker> workers = new HashSet<Worker>(); //正在被执行的Worker任务集

3.1.3 线程池的主要状态锁

　　线程池内部的状态变化 ( 比如线程池大小、runState等 ) 都需要基于此锁。

private final ReentrantLock mainLock = new ReentrantLock();

3.1.4 线程的存活时间和大小

1 private volatile long keepAliveTime;// 线程存活时间
2 private volatile boolean allowCoreThreadTimeOut;// 是否允许核心线程存活
3 private volatile int corePoolSize;// 核心池大小
4 private volatile int maximumPoolSize; // 最大池大小
5 private volatile int poolSize; //当前池大小
6 private int largestPoolSize; //最大池大小，区别于maximumPoolSize，是用于记录线程池曾经达到过的最大并发,理论上小于等于maximumPoolSize。

3.1.5 线程工厂和拒绝策略

1 private volatile RejectedExecutionHandler handler;// 拒绝策略，用于当线程池无法承载新线程是的处理策略。
2  private volatile ThreadFactory threadFactory;// 线程工厂，用于在线程池需要新创建线程的时候创建线程

3.1.6 线程池完成任务数和最大线程数

1 private long completedTaskCount;//线程池运行到当前完成的任务数总和
2 private int largestPoolSize;   //用来记录线程池中曾经出现过的最大线程数

　　对于corePoolSize、maximumPoolSize、largestPoolSize变量可借助与下面的例子帮助加深理解：

　　假如有一个工厂，工厂里面有10个工人，每个工人同时只能做一件任务。因此只要当10个工人中有工人是空闲的，来了任务就分配给空闲的工人做；当10个工人都有任务在做时，如果还来了任务，就把任务进行排队等待；如果说新任务数目增长的速度远远大于工人做任务的速度，那么此时工厂主管可能会想补救措施，比如重新招4个临时工人进来；然后就将任务也分配给这4个临时工人做；如果说着14个工人做任务的速度还是不够，此时工厂主管可能就要考虑不再接收新的任务或者抛弃前面的一些任务了。当这14个工人当中有人空闲时，而新任务增长的速度又比较缓慢，工厂主管可能就考虑辞掉4个临时工了，只保持原来的10个工人，毕竟请额外的工人是要花钱的。

　　这个例子中的corePoolSize就是10，而maximumPoolSize就是14（10+4）。也就是说corePoolSize就是线程池大小，maximumPoolSize在我看来是线程池的一种补救措施，即任务量突然过大时的一种补救措施。largestPoolSize只是一个用来起记录作用的变量，用来记录线程池中曾经有过的最大线程数目，跟线程池的容量没有任何关系。

3.2 ThreadPoolExecutor的内部工作原理

ThreadPoolExecutor的内部工作原理总结起来就是 5 句话：

如果当前池大小 poolSize 小于 corePoolSize ，则创建新线程执行任务。
如果当前池大小 poolSize 大于 corePoolSize ，且等待队列未满，则进入等待队列
如果当前池大小 poolSize 大于 corePoolSize 且小于 maximumPoolSize ，且等待队列已满，则创建新线程执行任务。
如果当前池大小 poolSize 大于 corePoolSize 且大于 maximumPoolSize ，且等待队列已满，则调用拒绝策略来处理该任务。
线程池里的每个线程执行完任务后不会立刻退出，而是会去检查下等待队列里是否还有线程任务需要执行，如果在 keepAliveTime 里等不到新的任务了，那么线程就会退出。

3.3 ThreadPoolExecutor源码分析

在ThreadPoolExecutor类中，最核心的任务提交方法是execute()方法，虽然通过submit也可以提交任务，但是实际上submit方法里面最终调用的还是execute()方法：

 1 public void execute(Runnable command) {
 2     if (command == null)
 3         throw new NullPointerException();
 4     if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {
 5         if (runState == RUNNING && workQueue.offer(command)) {
 6             if (runState != RUNNING || poolSize == 0)
 7                 ensureQueuedTaskHandled(command);
 8         }
 9         else if (!addIfUnderMaximumPoolSize(command))
10             reject(command); // is shutdown or saturated
11     }
12 }

　　一个任务通过 execute(Runnable)方法被添加到线程池，任务就是一个Runnable类型的对象，任务的执行方法就是run()方法，如果传入的为null，侧抛出NullPointerException。
　　如果当前线程数小于corePoolSize，调用addIfUnderCorePoolSize方法，addIfUnderCorePoolSize方法首先调用mainLock加锁，再次判断当前线程数小于corePoolSize并且线程池处于RUNNING状态，则调用addThread增加线程
addIfUnderCorePoolSize方法实现：

 1 private boolean addIfUnderCorePoolSize(Runnable firstTask) {
 2     Thread t = null;
 3     final ReentrantLock mainLock = this.mainLock;
 4     mainLock.lock();
 5     try {
 6         if (poolSize < corePoolSize && runState == RUNNING)
 7             t = addThread(firstTask);        //创建线程去执行firstTask任务
 8         } finally {
 9         mainLock.unlock();
10     }
11     if (t == null)
12         return false;
13     t.start();
14     return true;
15 }

　　addThread方法首先创建Work对象，然后调用threadFactory创建新的线程，如果创建的线程不为null，将Work对象的thread属性设置为此创建出来的线程，并将此Work对象放入workers中，然后在增加当前线程池的中线程数，增加后回到addIfUnderCorePoolSize方法，释放mainLock，最后启动这个新创建的线程来执行新传入的任务。
addThread方法实现：

 1 private Thread addThread(Runnable firstTask) {
 2     Worker w = new Worker(firstTask);
 3     Thread t = threadFactory.newThread(w);  //创建一个线程，执行任务
 4     if (t != null) {
 5         w.thread = t;            //将创建的线程的引用赋值为w的成员变量
 6         workers.add(w);
 7         int nt = ++poolSize;     //当前线程数加1
 8         if (nt > largestPoolSize)
 9             largestPoolSize = nt;
10     }
11     return t;
12 }

　　从addThread方法看得出，Worker对象包装了参数传入的任务，threadFactory新创建的线程包装了Worker对象，在执行新创建线程的run方法时，调用到了Worker对象的run方法。

Worker类最核心的run方法如下：

 1 public void run() {
 2     try {
 3         Runnable task = firstTask;
 4         firstTask = null;
 5         while (task != null || (task = getTask()) != null) {
 6             runTask(task);
 7             task = null;
 8         }
 9     } finally {
10         workerDone(this);
11     }
12 }

　　从以上方法可以看出，Worker所在的线程启动后，首先执行创建其时传入的Runnable任务，执行完成后，循环调用getTask从任务缓存队列里面去获取新的任务，在没有任务的情况下，退出此线程。

getTask方法的实现如下：

 1 Runnable getTask() {
 2     for (;;) {
 3         try {
 4             int state = runState;
 5             if (state > SHUTDOWN)
 6                 return null;
 7             Runnable r;
 8             if (state == SHUTDOWN)  // Help drain queue
 9                 r = workQueue.poll();
10             else if (poolSize > corePoolSize || allowCoreThreadTimeOut) //如果线程数大于核心池大小或者允许为核心池线程设置空闲时间，
11                                                                         //则通过poll取任务，若等待一定的时间取不到任务，则返回null。
12                 r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS);
13             else
14                 r = workQueue.take();
15             if (r != null)
16                 return r;
17             if (workerCanExit()) {    //如果没取到任务，即r为null，则判断当前的worker是否可以退出
18                 if (runState >= SHUTDOWN) // Wake up others
19                     interruptIdleWorkers();   //中断处于空闲状态的worker
20                 return null;
21             }
22             // Else retry
23         } catch (InterruptedException ie) {
24             // On interruption, re-check runState
25         }
26     }
27 }

　　getTask就是通过WorkQueue的poll或task方法来获取下一个要执行的任务。它先判断当前线程池状态，如果runState大于SHUTDOWN（即为STOP或者TERMINATED），则直接返回null。如果runState为SHUTDOWN或者RUNNING，则从任务缓存队列取任务。

　　回到execute方法代码的5-10行。如果当前线程池数量大于corePoolSize或addIfUnderCorePoolSize方法执行失败，则执行后续操作；如果线程池处于运行状态并且workQueue中成功加入任务，再次判断如果线程池的状态不为运行状态或当前线程池数为0，则调用ensureQueuedTaskHandled方法。

 1 private void ensureQueuedTaskHandled(Runnable command) {
 2     final ReentrantLock mainLock = this.mainLock;
 3     mainLock.lock();
 4     boolean reject = false;
 5     Thread t = null;
 6     try {
 7         int state = runState;
 8         if (state != RUNNING && workQueue.remove(command))
 9             reject = true;
10         else if (state < STOP &&
11                  poolSize < Math.max(corePoolSize, 1) &&
12                  !workQueue.isEmpty())
13             t = addThread(null);
14     } finally {
15         mainLock.unlock();
16     }
17     if (reject)
18         reject(command);
19     else if (t != null)
20         t.start();
21 }

　　ensureQueuedTaskHandled方法判断线程池运行，如果状态不为运行状态，从workQueue中删除，并调用reject做拒绝处理。

1 void reject(Runnable command) {
2     handler.rejectedExecution(command, this);
3 }

　　再次回到execute方法代码的5-10行。如线程池workQueue offer失败或不处于运行状态，调用addIfUnderMaximumPoolSize，addIfUnderMaximumPoolSize方法基本和addIfUnderCorePoolSize实现类似，不同点在于根据最大线程数（maximumPoolSize）进行比较，如果超过最大线程数，返回false，调用reject方法。

3.4 线程池中的线程初始化

　　默认情况下，创建线程池之后，线程池中是没有线程的，需要提交任务之后才会创建线程。在实际中如果需要线程池创建之后立即创建线程，可以通过以下两个方法办到：

prestartCoreThread()：初始化一个核心线程；
prestartAllCoreThreads()：初始化所有核心线程

3.5 任务拒绝策略

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

1 ThreadPoolExecutor.AbortPolicy         //丢弃任务并抛出RejectedExecutionException异常。
2 ThreadPoolExecutor.DiscardPolicy       //也是丢弃任务，但是不抛出异常。
3 ThreadPoolExecutor.DiscardOldestPolicy //丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程）
4 ThreadPoolExecutor.CallerRunsPolicy    //由调用线程处理该任务

3.6 线程池的关闭

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

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

四. 配置线程池的大小

　　一般需要根据任务的类型来配置线程池大小：

　　如果是CPU密集型任务，就需要尽量压榨CPU，参考值可以设为 NCPU+1
　　如果是IO密集型任务，参考值可以设置为2*NCPU

　　当然，这只是一个参考值，具体的设置还需要根据实际情况进行调整，比如可以先将线程池大小设置为参考值，再观察任务运行情况和系统负载、资源利用率来进行适当调整。

参考：http://www.cnblogs.com/dolphin0520/p/3932921.html

http://blog.csdn.net/java2000_wl/article/details/22097059

http://cuisuqiang.iteye.com/blog/2019372

http://developer.51cto.com/art/201203/321885.htm

时间： 2024-10-17 11:27:05

Java并发之——线程池