ThreadPoolExecutor机制&线程池详解

Posted 2021-01-28 yuanfei1110111

一、概述
1.线程池可以解决两个不同问题：由于减少了每个任务调用的开销，它们通常可以在执行大量异步任务时提供增强的性能，并且还可以提供绑定和管理资源（包括执行任务集时使用的线程）的方法。每个 ThreadPoolExecutor 还维护着一些基本的统计数据，如完成的任务数；
2.ThreadPoolExecutor作为java.util.concurrent包对外提供的基础实现，以内部线程池的形式对外提供管理任务执行、线程调度、线程池管理等等服务；
3.Executors方法提供的线程服务，都是通过参数设置来实现不同的线程池机制。如 Executors.newCachedThreadPool()（无界线程池，可以进行自动线程回收）、Executors.newFixedThreadPool(int)（固定大小线程池）和 Executors.newSingleThreadExecutor()（单个后台线程），它们均为大多数使用场景预定义了设置。

二、核心构造方法参数讲解
corePoolSize：核心线程池大小；
maximumPoolSize：最大线程池大小；
keepAliveTime：线程池中超过corePoolSize数目的空闲线程最大存活时间；
TimeUnit：keepAliveTime的时间单位；
workQueue：阻塞任务队列；
threadFactory：新建线程工厂；
RejectedExecutionHandler：当提交任务数量超过maximumPoolSize+workQueue之和时，任务会交给RejectedExecutionHandler处理。

重点讲解：
其中比较容易让人误解的是：corePoolSize、maximumPoolSize、workQueue之间的关系：
1.当线程池小于corePoolSize时，新提交任务将创建一个新线程执行任务，即使此时线程池中存在空闲线程；
2.当线程池达到corePoolSize时，新提交任务将被放入workQueue中，等待线程池中任务调度执行；
3.当workQueue已满，且maximumPoolSize>CorePoolSize时，新提交的任务会创建新线程执行任务；
4.当提交任务数超过maximumPoolSize时，新提交任务由RejectedExecutionHandler处理；
5.当线程池中超过corePoolSize线程，空闲时间达到keepAliveTime时，关闭空闲线程；
6.当设置allowCoreThreadTimeOut(true)时，线程池中corePoolSize线程空闲时间达到keepAliveTime也将关闭。

三、Executors提供的线程池配置方案
1.构造一个固定线程数目的线程池，配置的corePoolSize与maximumPoolSize大小相同，同时使用了一个无界的LinkedBlockingQueue存放阻塞任务，因此多余的任务将存在在阻塞队列，不会由RejectedExecutionHandler处理：
    public static ExecutorService newFixedThreadPool(int nThreads) {  
            return new ThreadPoolExecutor(nThreads, nThreads,  
                                          0L, TimeUnit.MILLISECONDS,  
                                          new LinkedBlockingQueue<Runnable>());  
        }
2.构造一个缓冲功能的线程池，配置corePoolSize=0，maximumPoolSize=Integer.MAX_VALUE，keepAliveTime=60s以及一个无容量的阻塞队列SynchronousQueue，因此任务提交之后，将会创建新的线程执行，线程空闲超过60s将会销毁：
    public static ExecutorService newCachedThreadPool() {  
            return new ThreadPoolExecutor(0, Integer.MAX_VALUE,  
                                          60L, TimeUnit.SECONDS,  
                                          new SynchronousQueue<Runnable>());  
        }  
3.构造一个只支持一个线程的线程池，配置corePoolSize=maximumPoolSize=1，无界阻塞队列LinkedBlockingQueue，保证任务由一个线程串行执行：
    public static ExecutorService newSingleThreadExecutor() {  
            return new FinalizableDelegatedExecutorService  
                (new ThreadPoolExecutor(1, 1,  
                                        0L, TimeUnit.MILLISECONDS,  
                                        new LinkedBlockingQueue<Runnable>()));  
        }  

四、自定义线程池的注意要点（相关代码见url）
1、用ThreadPoolExecutor自定义线程池，看线程是的用途，如果任务量不大，可以用无界队列，如果任务量非常大，要用有界队列，防止OOM；
2、如果任务量很大，还要求每个任务都处理成功，要对提交的任务进行阻塞提交，重写拒绝机制，改为阻塞提交。保证不抛弃一个任务；
3、最大线程数一般设为2N+1最好，N是CPU核数；
4、核心线程数，看应用，如果是任务，一天跑一次，设置为0，合适，因为跑完就停掉了，如果是常用线程池，看任务量，是保留一个核心还是几个核心线程数；
5、如果要获取任务执行结果，用CompletionService，但是注意，获取任务的结果的要重新开一个线程获取，如果在主线程获取，就要等任务都提交后才获取，就会阻塞大量任务结果，队列过大OOM，所以最好异步开个线程获取结果。

五、线程池状态含义：
1.running：接收新任务并处理阻塞队列里的任务；
2.shutdown：拒绝新任务但是处理阻塞队列里的任务；
3.stop：拒绝新任务并且抛弃阻塞队列里的任务同时会中断正在处理的任务；
4.tidying：所有任务都执行完（包括阻塞队列里的任务），当线程池里活动线程为0，将调用terminated方法；
5.terminated：终止状态，terminated方法调用完成以后的状态。

六、按需构造
默认情况下，即使核心线程最初只是在新任务到达时才创建和启动的，也可以使用方法 prestartCoreThread() 或 prestartAllCoreThreads() 对其进行动态重写。如果构造带有非空队列的池，则可能希望预先启动线程。

七、创建新线程
使用 ThreadFactory 创建新线程。如果没有另外说明，则在同一个 ThreadGroup 中一律使用 Executors.defaultThreadFactory() 创建线程，并且这些线程具有相同的 NORM_PRIORITY 优先级和非守护进程状态。通过提供不同的 ThreadFactory，可以改变线程的名称、线程组、优先级、守护进程状态，等等。如果从 newThread 返回 null 时 ThreadFactory 未能创建线程，则执行程序将继续运行，但不能执行任何任务。

八、排队
所有 BlockingQueue 都可用于传输和保持提交的任务。可以使用此队列与池大小进行交互：
1.如果运行的线程少于 corePoolSize，则 Executor 始终首选添加新的线程，而不进行排队。
2.如果运行的线程等于或多于 corePoolSize，则 Executor 始终首选将请求加入队列，而不添加新的线程。
3.如果无法将请求加入队列，则创建新的线程，除非创建此线程超出 maximumPoolSize，在这种情况下，任务将被拒绝。

九、排队的三种通用策略
1.直接提交：工作队列的默认选项是 SynchronousQueue，它将任务直接提交给线程而不保持它们（没有容量）。在此，如果不存在可用于立即运行任务的线程，则试图把任务加入队列将失败，因此会构造一个新的线程。此策略可以避免在处理可能具有内部依赖性的请求集时出现锁。直接提交通常要求无界 maximumPoolSizes 以避免拒绝新提交的任务。当命令以超过队列所能处理的平均数连续到达时，此策略允许无界线程具有增长的可能性。
2.无界队列：使用无界队列（例如，不具有预定义容量的 LinkedBlockingQueue）将导致在所有 corePoolSize 线程都忙时新任务在队列中等待。这样，创建的线程就不会超过 corePoolSize。（因此，maximumPoolSize 的值也就无效了。）当每个任务完全独立于其他任务，即任务执行互不影响时，适合于使用无界队列；例如，在 Web 页服务器中。这种排队可用于处理瞬态突发请求，当命令以超过队列所能处理的平均数连续到达时，此策略允许无界线程具有增长的可能性。
3.有界队列：当使用有限的 maximumPoolSizes 时，有界队列（如 ArrayBlockingQueue）有助于防止资源耗尽，但是可能较难调整和控制。队列大小和最大池大小可能需要相互折衷：使用大型队列和小型池可以最大限度地降低 CPU 使用率、操作系统资源和上下文切换开销，但是可能导致人工降低吞吐量。如果任务频繁阻塞（例如，如果它们是 I/O 边界），则系统可能为超过您许可的更多线程安排时间。使用小型队列通常要求较大的池大小，CPU 使用率较高，但是可能遇到不可接受的调度开销，这样也会降低吞吐量。

十、被拒绝的任务
当 Executor 已经关闭，或者 Executor 将有限边界用于最大线程和工作队列容量，且已经饱和时，在方法 execute(java.lang.Runnable) 中提交的新任务将被拒绝。在以上两种情况下，execute 方法都将调用其 RejectedExecutionHandler 的 RejectedExecutionHandler.rejectedExecution(java.lang.Runnable, java.util.concurrent.ThreadPoolExecutor) 方法。下面提供了四种预定义的处理程序策略：
1.在默认的 ThreadPoolExecutor.AbortPolicy 中，处理程序遭到拒绝将抛出运行时 RejectedExecutionException。
2.在 ThreadPoolExecutor.CallerRunsPolicy 中，线程调用运行该任务的 execute 本身。此策略提供简单的反馈控制机制，能够减缓新任务的提交速度。
3.在 ThreadPoolExecutor.DiscardPolicy 中，不能执行的任务将被删除。
4.在 ThreadPoolExecutor.DiscardOldestPolicy 中，如果执行程序尚未关闭，则位于工作队列头部的任务将被删除，然后重试执行程序（如果再次失败，则重复此过程）。定义和使用其他种类的 RejectedExecutionHandler 类也是可能的，但这样做需要非常小心，尤其是当策略仅用于特定容量或排队策略时。

十一 hook（钩子）方法
ThreadPoolExecutor提供 protected 可重写的 beforeExecute(java.lang.Thread, java.lang.Runnable) 和 afterExecute(java.lang.Runnable, java.lang.Throwable) 方法，这两种方法分别在执行每个任务之前和之后调用。它们可用于操纵执行环境；例如，重新初始化 ThreadLocal、搜集统计信息或添加日志条目。此外，还可以重写方法 terminated() 来执行 Executor 完全终止后需要完成的所有特殊处理。如果钩子 (hook) 或回调方法抛出异常，则内部辅助线程将依次失败并突然终止。

十二 队列维护
方法 getQueue() 允许出于监控和调试目的而访问工作队列。强烈反对出于其他任何目的而使用此方法。remove(java.lang.Runnable) 和 purge() 这两种方法可用于在取消大量已排队任务时帮助进行存储回收。

十三 终止
程序 AND 不再引用的池没有剩余线程会自动 shutdown。如果希望确保回收取消引用的池（即使用户忘记调用 shutdown()），则必须安排未使用的线程最终终止：设置适当保持活动时间，使用 0 核心线程的下边界和/或设置 allowCoreThreadTimeOut(boolean)。