zoukankan      html  css  js  c++  java

  • ThreadPoolExecutor详解

    一、源码分析（基于JDK1.6）

    　　ThreadExecutorPool是使用最多的线程池组件，了解它的原始资料最好是从从设计者（Doug Lea）的口中知道它的来龙去脉。在Jdk1.6中，ThreadPoolExecutor直接继承了AbstractExecutorService, 并层级实现了ExecutorService和Executor接口。

    1.Executor

    　　Executor是用来执行提交的Runnable任务的对象，并以接口的形式定义，提供一种提交任务(submission task）与执行任务(run task)之间的解耦方式，还包含有线程使用与周期调度的详细细节等。Executor常常用来代替早期的线程创建方式，如new Thread(new(RunnableTask())).start(),在实际中可以用如下的方式来提交任务到线程池里，Executor会自动执行 你的任务.

    1. Executor executor = anExecutor;  
    2. executor.execute(new RunnableTask1());  

    　　Executor接口中定义的方法如下：

    1. /** 
    2. 在接下来的某个时刻执行提交的command任务。由于Executor不同的实现，执行的时候可能在一个新线程中或由一个线程池里的线程执行，还可以是由调用者线程执行 
    3. */  
    4. void execute(Runnable command);  

    　　2.ExecutorService

    　　ExecutorService接口扩展了Executor,提供管理线程池终止的一组方法，还提供了产生Future的方法，Future是用于追踪一个或多个异步任务的对象，并能返回异步任务的计算结果。

    　　ExecutorService关闭后将不再接收新的任务，ExecutorService提供了两种不同类型的关闭方法，shutdown方法允许执行完之前提交的任务才终止，而shutdownNow将不再执行等待的任务，并试图终止当前执行的任务。ExecutorService终止后，内部已没有活动的任务，没有等待的任务，也不能再提交新任务，没有使用ExecutorService需要回收相应的资源。

    　　submit方法是基于Executor.execute()方法之上的，通过创建并返回一个Future对象就可以实现取消执行或者等待执行完成。invokeAny和invokeAll通常是用于批量执行，可以提交一个task集合并等待task的逐个完成。

    　　ExecutorService接口中定义的方法如下：

    1. //不再接收新的task,执行完之前提交的task后，开始有序的终止线程。 
    2. void shutdown();  
    3. /** 
    4. 试图终止所有活动的执行任务，停止对等待任务的处理，并返回待执行的Runnable列表。 
    5. 但是，不能保证能够终止掉所有的正在执行的任务。比如，在典型的实现中会调用Thread.interupt来作取消，但是一些不能响应中断的task将永远不会被终止 
    6. */  
    7. List<Runnable> shutdownNow();  
    8. //如果Executor调用shutdown或者shutdownNow将返回true
    9. boolean isShutdown(); 
    10. //所有的任务都关闭后，线程池才会关闭成功。届时返回true
    11. boolean isTerminated(); 
    12. /** 
    13. 发起关闭请求后，将一直阻塞等待关闭，直到所有的task已执行完成。 
    14. 如果超时返回，或者当前线程中断时， 则返回false 
    15. */  
    16. boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException;  
    17. //提交一个等待返回值的task,返回的Future表示task执行后的待定结果。执行成功后，Future的get方法将返回实际的结果。
    18. <T> Future<T> submit(Callable<T> task);  
    19. <T> Future<T> submit(Runnable task, T result);  
    20. //提交一个Runnable任务并返回一个Future。不过Future.get方法将返回null
    21. Future<?> submit(Runnable task);  
    22. // 执行提交的task集合。当执行完成后，返回task各自的Future。对应返回的Future集合，Future.isDone方法将返回true
    23. <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>>tasks)throws InterruptedException;  
    24. //执行提交的task集合，返回一个task成功执行后的结果,而其他没有执行完成的task将被取消
    25. <T> T invokeAny(Collection<? extends Callable<T>> tasks)throws InterruptedException, ExecutionException;     

    　　3. ThreadPoolExecutor

    　　ThreadPoolExecutor使用线程池里的线程来执行每个提交的task,一般通过Executors的工厂方法来创建并配置相应的参数。

    　　线程池处理了两个问题：1.改善了性能。当执行大量的异步任务时，减少了每次调用的开销。2.提供了一直资源管理的方式。线程池内部的线程可以得到有效的复用。每个ThreadPoolExecutor内部还维护了一些基础的统计量，如完成的任务总数.ThreadPoolExecutor在很多场景下都有其用武之地，它提供了很多可调整的参数与可扩展的钩子方法。一般建议用Executors提供的便利的工厂方法来创建相应的ThreadPoolExecutor。如可以创建无限多线程newCachedThreadPool方法，创建固定大小的newFixedThreadPool方法，还有创建单个后台线程newSingleThreadExecutor的方法。这些预定义的方法能满足大 分的使用场景，然而当需要手动配置与调整线程池时，则需要知晓内部的究竟。

    　　Ø 核心线程池大小(corePoolSize)与最大线程池大小(maximumPoolSize)

    　　ThreadPoolExecutor会自动调节池子里线程的大小。通过execute方法提交新任务时，如果当前的池子里线程的大小小于核心线程 corePoolSize时，则会创建新线程来处理新的请求，即使当前的工作者线程是空闲的。如果运行的线程数是大于corePoolSize但小于 maximumPoolSize，而且当任务队列已经满了时，则会创建新线程。通过设置corePoolSize等于maximumPoolSize，便 可以创建固定大小的线程池数量。而设置maximumPoolSize为一个不受限制的数量如Integer.MAX，便可以创建一个适应任意数量大的并发任务的线程池。常规情况下，可以根据构造方法来设置corePoolSize与maximumPoolSize，但也可以通过 setCorePoolSize和setMaximumPoolSize方法动态的修改其值。

    　　Ø 按需构造

    　　默认情况下，核心线程是在开始接收新任务时才初始创建，但是可以使用prestartCoreThread或prestartAllCoreThreads方法来动态的预开启所有的线程数。

    　　Ø  创建新线程

    　　新线程是使用java.util.concurrent.ThreadFactory来创建的，如果没有指定其他的方式，则是使用Executors.defaultThreadFactory方法，默认创建的线程拥有相同线程组与优先级且都是非后台线程。

    　　Ø  存活时间（Keep-alive times）

    　　若当前池里的线程数量超过corePoolSize，超出的线程如果是空闲的，将在存活指定的keepAliveTime时间后终止。这种机制主要是为减少资源的消耗，如果后期有新的活动任务，则又构造新的线程。该参数也可以通过setKeepAliveTime方法动态的修改，如果该参数设置为 Long.MAX_VALUE，则空闲的线程将一直存活。

    　　Ø  阻塞队列

    　　超出一定数量的任务会转移队列中，队列与池里的线程大小的关联表现在：如运行的线程数小于corePoolSize线程数，Executor会优先添加线程来执行task,而不会添加到队列中。如运行的线程已大于 corePoolSize,Executor会把新的任务放于队列中，如队列已到最大时，ThreadPoolExecutor会继续创建线程，直到超过 maximumPoolSize。最后，线程超过maximumPoolSize时，Executor将拒绝接收新的task.

    　　而添加任务到队列时，有三种常规的策略：

    1. 直接传递。SynchronousQueue队列的默认方式，一个存储元素的阻塞队列而是直接投递到线程中。每一个入队操作必须等到另一个线程调用移除操作，否则入队将一直阻塞。当处理一些可能有内部依赖的任务时，这种策略避免了加锁操作。直接传递一般不能限制maximumPoolSizes以避免拒绝 接收新的任务。如果新增任务的速度大于任务处理的速度就会造成增加无限多的线程的可能性。

    2. 无界队列。如LinkedBlockingQueue，当核心线程正在工作时，使用不用预先定义大小的无界队列将使新到来的任务处理等到中，所以如果线程数是小于corePoolSize时，将不会创建有入队操作。这种策略将很适合那些相互独立的任务，如Web服务器。如果新增任务的速度大于任务处理的速度就会造成无界队列一直增长的可能性。

    3. 有界队列。如ArrayBlockingQueue，当定义了maximumPoolSizes时使用有界队列可以预防资源的耗尽，但是增加了调整和控制队列的难度，队列的大小和线程池的大小是相互影响的，使用很大的队列和较小的线程池会减少CPU消耗、操作系统资源以及线程上下文开销，但却人为的降低了吞吐量。如果任务是频繁阻塞型的（I/O），系统是可以把时间片分给多个线程的。而采用较小的队列和较大的线程池，虽会造成CPU繁忙，但却会遇到调度开销，这也会降低吞吐量。

    　　Ø  饱和策略（拒绝接收任务）

    　　当Executor调用shutdown方法后或者达到工作队列的最容量时,线程池则已经饱和了，此时则不会接收新的task。但无论是何种情 况，execute方法会调用RejectedExecutionHandler#rejectedExecution方法来执行饱和策略，在线程池内部预定义了几种处理策略：

    1. 终止执行(AbortPolicy)。默认策略， Executor会抛出一个RejectedExecutionException运行异常到调用者线程来完成终止。

    2. 调用者线程来运行任务(CallerRunsPolicy)。这种策略会由调用execute方法的线程自身来执行任务，它提供了一个简单的反馈机制并能降低新任务的提交频率。

    3. 丢弃策略(DiscardPolicy)。不处理，直接丢弃提交的任务。

    4. 丢弃队列里最近的一个任务(DiscardOldestPolicy)。如果Executor还未shutdown的话，则丢弃工作队列的最近的一个任务，然后执行当前任务。

    　　终于可以走近代码了，在ThreadPoolExecutor，我们直接跳跃到核心的字段和方法处。

    　　ThreadPoolExecutor类分析：

    1. /* 
    2. 一 个ThreadPoolExecutor管理着一系列控制字段。首先需要保证在加锁的区域中， 执行的控制字段才能被改变，这些字段包含有 runState，poolSize, corePoolSize, and maximumPoolSize。然后这些字段被声明为volatile类 型，因此保证了内存可见性(任何线程都有对其的内存可见性)。 
    3. 而一些其他的字段是表示用户控制的参数，不会影响执行的结果，也声明为volatile类型，允许用户在执行时异步的改变。这些字段有包含有：allowCoreThreadTimeOut, keepAliveTime。除此之外，内部的饱和策略处理器、线程工厂也不会在加锁区域内被更 改。 
    4. 大量的volatile类型声明主要是保证在不用加锁的条件下，很多操作的结果都具有内存的可见性，如工作队列的任务入队和出队操作。 
    5. */  
    6. /** 
    7. runState提供生命周期的控制，有如下值： 
    8. RUNNING: 接收新任务并正在队列中的任务 
    9. SHUTDOWN: 不再接收新的任务，但是仍在处理队列中的任务 
    10. STOP：不再接收新的任务，不再处理队列中的任务并中断正在执行的任务 
    11. TERMINATED：同STOP一样，同时所有的线程已被终止 
    12. runState会单一的增加，但不需要每个值都命中，他们可有如下的转换顺序： 
    13. RUNNING -> SHUTDOWN 调用shutdown方法后 
    14. (RUNNING or SHUTDOWN) -> STOP 调用shutdownNow()后 
    15. SHUTDOWN -> TERMINATED 队列和线程池里的任务已完，线程已终止 
    16. STOP -> TERMINATED 线程池已空，线程已终止 
    17. */  
    18. volatile int runState;  
    19. static final int RUNNING= 0;  
    20. static final int SHUTDOWN= 1;  
    21. static final int STOP = 2;  
    22. static final int TERMINATED = 3;
    23. //用于存储任务并把任务投递给工作者线程的阻塞队列。我们不需用workQueue.poll()方法返回null,因为此时可认为队列已为空，因此有时候必须再作检查
    24. private final BlockingQueue<Runnable> workQueue;
    25. /** 
    26. 更新poolSize, corePoolSize, maximumPoolSize, runState, and workers者线程的锁 
    27. 在1.6中是一把重入锁，在1.7时已经修改为直接继承AQS。 
    28. */  
    29. private final ReentrantLock mainLock = new ReentrantLock();  
    30. //用于辅助awaitTermination方法的等待条件
    31. private final Condition termination = mainLock.newCondition();  
    32. //池里的工作者线程，当持有mainLock时才能读取 
    33. private final HashSet<Worker> workers = new HashSet<Worker>();  
    34. // 以纳秒为单位的超时时间，如果allowCoreThreadTimeOut设置为true,则空闲的多余corePoolSize的线程将会在存活keepAliveTime时长后终止。
    35. private volatile long  keepAliveTime;  
    36. // 默认为false, 保留空闲的核心线程
    37. private volatile boolean allowCoreThreadTimeOut;  
    38. private volatile int   corePoolSize;  
    39. private volatile int   maximumPoolSize;  
    40. private volatile int   poolSize;  
    41. //饱和执行策略或者shutdown后拒绝执行的策略 
    42. private volatile RejectedExecutionHandler handler;  
    43. //创建新线程的工厂。 
    44. private volatile ThreadFactory threadFactory;  

    　　接下来便是一组按需指定的构造方法

    　　分析一下最重要的execute方法。
    　　1. 如果当前池里运行的线程数量小于corePoolSize,则创建新线程（需要获取全局锁） 　　2. 如果当前的线程数量大于corePoolSize，则将任务加入BlockQueue中。 　　3. 如果队列已满，但是线程数小于最大线程数量，则继续添加线程（需要再次获取全局锁） 　　4. 已达到最大线程数量，任务队列也已经满了，任务将被拒绝，并调用RejectExecutionHanlder的rejectExecution方法。 　　ThreadPoolExecutor采用上述步骤的总体思路为，在执行execute方法时，尽可能的避免获取全局锁（影响性能），在 　 ThreadPoolExecutor完成核心线程的开启后，几乎所有的execute方法都是执行步骤2，巧妙地是，步骤2不需要获取全局锁。

    1. public void execute(Runnable command) {  
    2.         if (command == null) //不能是空任务  
    3.             throw new NullPointerException();  
    4.     //如果还没有达到corePoolSize，则添加新线程来执行任务  
    5.         if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {  
    6.          //如果已经达到corePoolSize,则不断的向工作队列中添加任务  
    7.             if (runState == RUNNING && workQueue.offer(command)) {  
    8.             //线程池已经没有任务  
    9.                 if (runState != RUNNING || poolSize == 0)   
    10.                     ensureQueuedTaskHandled(command);  
    11.             }  
    12.          //如果线程池不处于运行中或者工作队列已经满了，但是当前的线程数量还小于允许最大的maximumPoolSize线程数量，则继续创建线程来执行任务  
    13.             else if (!addIfUnderMaximumPoolSize(command))  
    14.             //已达到最大线程数量，任务队列也已经满了，则调用饱和策略执行处理器  
    15.                 reject(command); // is shutdown or saturated  
    16.         }  
    17. }  
    18.   
    19. private boolean addIfUnderCorePoolSize(Runnable firstTask) {  
    20.         Thread t = null;  
    21.         final ReentrantLock mainLock = this.mainLock;  
    22.         mainLock.lock();  
    23.         //更改几个重要的控制字段需要加锁  
    24.         try {  
    25.             //池里线程数量小于核心线程数量，并且还需要是运行时  
    26.             if (poolSize < corePoolSize && runState == RUNNING)  
    27.                 t = addThread(firstTask);  
    28.         } finally {  
    29.             mainLock.unlock();  
    30.         }  
    31.         if (t == null)  
    32.             return false;  
    33.         t.start(); //创建后，立即执行该任务  
    34.         return true;  
    35.     }  
    36.   
    37. private Thread addThread(Runnable firstTask) {  
    38.         Worker w = new Worker(firstTask);  
    39.         Thread t = threadFactory.newThread(w); //委托线程工厂来创建，具有相同的组、优先级、都是非后台线程  
    40.         if (t != null) {  
    41.             w.thread = t;  
    42.             workers.add(w); //加入到工作者线程集合里  
    43.             int nt = ++poolSize;  
    44.             if (nt > largestPoolSize)  
    45.                 largestPoolSize = nt;  
    46.         }  
    47.         return t;  
    48.     }  

    　　再看看工作者线程的设计。

    　　线程池内部可以直接用一个线程集合来复用线程，但Doug先生用Worker再次封装一下，估计其设计的初衷有：划分职责，只做单纯的任务消费者，执行完成后可追寻一些统计量。Worker执行任务时，需要先获取runLock，此处的目的是在任务的执行过程中防止worker线程被中断。然后双重检查是否线程池已停止或者中断。最好开始执行任务，此时调用用户自定的钩子方法，可在执行前和执行后作相应的处理。在Worker自身的run方法体中，需要先获取任务，调用实际的runTask。在获取任务的操作中，由于是阻塞获取，则保证了线程的最终存活的可能。

    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.         }  

    最好看看从队列中获取任务的方法。

    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(); //poll方法不会阻塞  
    10.             //keepAliveTime的超时终止体现在此处，超时后poll方法会返回r.然后回到worker的run方法里，由于没有可用的任务，超时返回的r会为null,因此worker的run方法会直接退出循环和导致线程结束。  
    11.                 else if (poolSize > corePoolSize || allowCoreThreadTimeOut)  
    12.                     r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS);  
    13.                 else  
    14.                     r = workQueue.take(); //阻塞等待  
    15.                 if (r != null)  
    16.                     return r;  
    17.                 if (workerCanExit()) {  
    18.                     if (runState >= SHUTDOWN) // Wake up others  
    19.                         interruptIdleWorkers();  
    20.                     return null;  
    21.                 }  
    22.                 // Else retry  
    23.             } catch (InterruptedException ie) {  
    24.                 // On interruption, re-check runState  
    25.             }  
    26.         }  
    27.     }  

    至于此，ThreadPoolExecutor的核心源码已经分析完成。

    二、再回头来分析原理

    执行示意图

    三、线程池的使用

    　　1. 通过构造方法来初始化
    1. public ThreadPoolExecutor(int corePoolSize,  
    2.                               int maximumPoolSize,  
    3.                               long keepAliveTime,  
    4.                               TimeUnit unit,  
    5.                               BlockingQueue<Runnable> workQueue)   

    　　2.或使用Executors的工厂方法

    1. public static ExecutorService newFixedThreadPool(int nThreads) {  
    2.         return new ThreadPoolExecutor(nThreads, nThreads,  
    3.                                       0L, TimeUnit.MILLISECONDS,  
    4.                                       new LinkedBlockingQueue<Runnable>());  
    5. public static ExecutorService newCachedThreadPool() {  
    6.         return new ThreadPoolExecutor(0, Integer.MAX_VALUE,  
    7.                                       60L, TimeUnit.SECONDS,  
    8.                                       new SynchronousQueue<Runnable>());  
    　　3. 调用execute方法，提交Runnable
    1. executorPool.execute(new TaskAction(taskId));  

    　四、之前的一些疑惑

    　　初步接触线程池时，有些问题也一直得不到解答，看完源码后，便得到了一下解答。

    　　1. 线程为什么能够一直存活？

    　　每个Woker是一个Runnable，同时会绑定到一个线程上。在执行Worker的run方法时，会去队列中获取任务，但是获取任务是阻塞的获取，如果没有则线程会一直等待，因此不会被终止。最终还是会转移到重入锁上并有内部同步器来完成执行阻塞的操作。

    　　2. 参数keepAliveTime的具体原理？

    　　在ThreadPoolExecutor的getTask方法中，如果当前池里线程的数量大于核心数量或者设置 allowCoreThreadTimeOut为true的话，则调用的是阻塞队列的 poll(long timeout,TimeUnit unit)方法，该方法等待指定的时间后会直接返回。worker线程会获取到返回的任务，如果为空的话，则退出循环，因此线程便结束了。
  • 相关阅读:
    python使用mysql数据库
    拆掉思维里的墙
    拆掉思维里的墙
    拆掉思维里的墙
    loadrunner通过web的post请求方法测接口 2
    loadrunner通过web的post请求方法测接口 2
    loadrunner通过web的post请求方法测接口 1
    luogu P1186 玛丽卡
    luogu P1855 榨取kkksc03
    暑期前复习
  • 原文地址：https://www.cnblogs.com/wxgblogs/p/5469208.html
Copyright © 2011-2022 走看看