ThreadPoolExecutor线程池

Posted 2020-11-19 fdzfd

一：类继承结构

                            继承关系

二：构造函数

                                                   构造函数

（1）线程池的大小除了显示的限制外，还可能由于其他资源上的约束而存在一些隐式限制。比如JDBC连接池。

（2）运行时间较长的任务。

    如果任务阻塞的时间过长，即使不出现死锁，线程池的响应性也会变得糟糕。执行时间较长的任务不仅会造成线程池阻塞，甚至还会增加执行时间。如果线程池中线程的数量远小于在稳定状态下执行时间较长任务的数量，那么到最后可能所有的线程都会运行这些执行时间较长的任务，从而影响整体的响应性。

    有一项技术可以缓解执行时间较长任务造成的影响，即限定任务等待资源的时间，而不要无限制地等待。在平台类库的大多数可阻塞方法中，都同时定义了限时版本和无限时版本，例如：Thread.join，BlockingQueue.put，CountDownLatch.await以及Selector.select等。如果等待超时，可以把任务标识为失败，然后中止任何或者将任务重新放回队列以便随后执行。如果在线程池中总是充满了呗阻塞的任务，那么也可能表示线程池的规模过小。

 

（3）设置线程池的大小

（3.1）线程池的理想大小取决于被提交任务的类型以及所部署系统的特性。在代码中不会固定线程池的大小，而应该通过某种配置机制来提供，或者根据Runtime.availableProcessors来动态计算。

（3.2）要设置线程池的大小也并不困难，只需要避免“过大”或“过小”这两种极端情况。如果设置过大，那么大量的线程将在相对很少的CPU和内存资源上发生竞争，这不仅会导致更高的内存使用量，而且还可能耗尽资源。如果设置过小，那么将导致很多空闲的处理器无法执行工作，从而降低吞吐率。

（3.3）要想正确设置线程池的大小，必须分析计算环境，资源预算和任务的特性。在部署的系统中有多少CPU？多大的内存？计算是计算密集型、I/O密集型还是二者皆可？它们是否需要像JDBC连接这样的稀缺资源？如果需要执行不同类别的任务，并且它们之间的行为相差很大，那么应该考虑使用多个线程池，从而使每个线程池可以根据自己的工作负载来调整。

（3.4）对于计算密集型的任务，在拥有Ncpu个处理器的系统上，当线程池的大小为Ncpu+1时，通常能实现最优的利用率。（计算当计算密集型的线程偶尔由于页缺失故障或者其他原因而暂停时，这个“额外”的线程也能确保CPU的时钟周期不会被浪费）

（3.5）对于包含I/O操作或者其他阻塞操作的任务，你必须估算出任务的等待时间与计算时间的比值。这种估算不需要很精确，并且可以通过一些分析或者监控工具来获得。你还可以通过另一种方法来调节线程池的大小：在某个基准负载下，分别设置不同大小的线程池来运行应用程序，并观察CPU利用率的水平。给定如下列定义：

                      Ncpu = number of CPUs

                     Ucpu = target CPU utilization，0 <= Ucpu <= 1

                     W/C = ratio of wait time to compute time

      要使处理器达到期望的使用率，线程池的最优大小等于：

                    Nthreads = Ncpu * Ucpu * (1 + W/C）

     可以通过Runtime来获得CPU的数目：

                   int N_CPUS = Runtime.getRuntime().availableProcessors()；

 

（4）参数解析

• corePoolSize

    线程池的基本大小（线程池的目标大小），默认情况下会一直存活，即使处于闲置状态也不会受keepAliveTime限制，除非将allowCoreThreadTimeOut设置为true。

• maximumPoolSize

    最大线程池大小，表示可同时活动的线程数量的上限。

• keepAliveTime

    如果某个线程的空闲时间超过了存活时间，那么将被标记为可回收的，并且当线程池的当前大小超过了基本大小时，这个线程将被终止。

分析：线程池的基本大小（corePoolSize）、最大大小（maximumPoolSize）以及存活时间等因素共同负责线程的创建与销毁。通过调节线程池的基本大小和存活时间，可以帮助线程池回收空闲线程占用的资源，从而使得这些资源可以用于执行其他工作。

 

三：基本实现

（1）newCachedThreadPool()

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

 

    线程池的基本大小设置为零，最大大小设置为Integer.MAX_VALUE，线程池可以被无限扩展，需求降低时自动收缩，最大大小设置过大在某些情况下也是缺点。

（2）newFixedThreadPool(int nThreads)

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

 

    缺点是LinkedBlockingQueue是无界队列，有些情况下排队的任务会很多。

（3）newScheduledThreadExecutor()

   

/**
     * Creates a thread pool that can schedule commands to run after a
     * given delay, or to execute periodically.
     * @param corePoolSize the number of threads to keep in the pool,
     * even if they are idle
     * @return a newly created scheduled thread pool
     * @throws IllegalArgumentException if {@code corePoolSize < 0}
     */
    public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
        return new ScheduledThreadPoolExecutor(corePoolSize);
    }

/**
     * Creates a new {@code ScheduledThreadPoolExecutor} with the
     * given core pool size.
     *
     * @param corePoolSize the number of threads to keep in the pool, even
     *        if they are idle, unless {@code allowCoreThreadTimeOut} is set
     * @throws IllegalArgumentException if {@code corePoolSize < 0}
     */
    public ScheduledThreadPoolExecutor(int corePoolSize) {
        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
              new DelayedWorkQueue());
    }

 

（4）newSingleThreadExecutor()

    

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

 

总结：都由Executors类的静态方法统一提供，如Executors.newCachedThreadPool()，底层通过ThreadPoolExecutor来实现。ThreadPoolExecutor提供了很多构造函数，它是一个灵活的、稳定的线程池，允许进行各种定制。

 

四：管理队列任务

（1）单线程的Executor是一种值得注意的特例：它们能确保不会有任务并发执行，因为它们通过线程封闭来实现线程安全性。

（2）如果无限制地创建线程，将会导致不稳定性。可以通过采用固定大小的线程池（而不是每收到一个请求就创建一个新线程）来解决这个问题。然而，这个方案并不完整。在高负载的情况下，应用程序仍可能耗尽资源，知识问题的概率较小。如果新请求的到达超过了线程池的处理效率，那么新到来的请求将累计起来。在线程池中，这些请求会在一个由Executor管理的Runable队列中等待，而不会像线程那样去竞争CPU资源，通过一个Runnable和一个链表节点来表现一个等待中的任务，当然比使用线程来表示的开销低得多，但如果客户提交给服务器请求的效率超过了服务器的处理效率，那么仍可能会耗尽资源。

（3）ThreadPoolExecutor允许提供一个BlockingQueue来保存等待执行的任务。基本的任务排队方法有3种：无界队列、有界队列和同步移交（Synchronous Handoff）。队列的选择与其他的配置参数有关，例如线程池的大小。

（4）newFixedThreadPool和newSingleThreadExecutor在默认的情况下将使用一个无界的LinkedBlockingQueue。如果所有工作者线程都处于忙碌状态，那么任务将在队列中等候。如果任务持续快速地到达，并且超过了线程池处理它们的速度，那么队列将无限制地增加。

（5）一种更稳妥的资源管理策略是使用有界队列，例如ArrayBlockingQueue、有界的LinkedBlockQueue、PriorityBlockingQueue。有界队列有助于避免资源耗尽的情况发生，但它又带来了新的问题：当队列填满后，新的任务该怎么办？（有许多饱和策略可以解决这个问题）在使用有界

的工作队列时，队列的大小和线程池的大小必须一起调节。如果线程池较小而队列较大，那么有助于减少内存使用量，降低CPU的使用率，同时还可以减少上下文切换，但付出的代价是可能会限制吞吐量。

（6）对于非常大的或者无界的线程池，可以通过使用SynchronousQueue来避免任务排队，以及直接将任务从生产者移交给工作者线程。SynchronousQueue不是一个真正的队列，而是一种在线程之间进行移交的机制。要将一个元素放入SynchronousQueue中，必须有另一个线程正在等待接受这个元素。

（7）当使用像LinkedBlockingQueue或ArrayBlockingQueue这样的FIFO（先进先出）队列时，任务的执行顺序与它们的到达顺序相同。如果想进一步控制任务执行顺序，还可以使用PriorityBlockingQueue，这个队列将根据优先级来安排任务。

（8）只有当任务相互独立时，为线程池或工作队列设置界限才是合理的。如果任务之间存在依赖性，那么有界的线程池或队列就可能导致线程“饥饿”死锁问题。此时应该使用无界的线程池，例如newCacheThreadPool。

 

五：饱和策略

（1）当有界队列被填满后，饱和策略开始发挥作用。ThreadPoolExecutor的饱和策略可以通过调用setRejectedExecutionHander来修改。（如果某个任务被提交到一个已被关闭的Executor时，也会用到饱和策略。）JDK提供了几种不同的RejectedExecutionHandler实现，每种实现都包含不同的饱和策略：AbortPolicy、CallerRunsPolicy、DiscardPolicy和DiscardOldestPolicy。

（2）中止（Abort）策略是默认的饱和策略，该策略将抛出未检查的RejectedExecutionException。调用者可以捕获这个异常，然后根据需要编写自己的处理代码。

（3）当新提交的任务无法保存到队列中等待执行时，“抛弃（Discard）”策略会悄悄抛弃该任务。

（4）“抛弃最旧的（Discard-Oldest）”策略则会抛弃下一个将被执行的任务，然后尝试重新提交新的任务。（如果工作队列是一个优先队列，那么“抛弃最旧的”策略将导致抛弃优先级最高的任务，因此最好不要将“抛弃最旧的”饱和策略和优先队列放在一起使用。）

（5）“调用者运行（Caller-Runs）”策略实现了一种调度机制，该策略既不会抛弃任务，也不会抛出异常，而是将某些任务回退到调用者，从而降低新任务的流量。它不会在线程池的某个线程中执行新提交的任务，而是在一个调用了execute的线程中执行该任务。如果采用有界队列和“调用者运行”饱和策略，当线程池中的所有线程都被占用，并且工作队列被填满后，下一个任务会在调用execute时在主线程中执行。由于执行任务需要一定的时间，因此主线程至少在一段时间内不能提交任何任务，从而使得工作者线程有时间来处理完正在执行的任务。在此期间，主线程不会调用accept，因此到达的请求将被保存在TCP层的队列中而不是在应用程序的队列中。如果持续过载，那么TCP层将最终发现它的请求队列被填满，因此同样会开始抛弃请求。当服务器过载时，这种过载情况会逐渐向外蔓延开来——从线程池到工作队列到应用程序再到TCP层，最终达到客户端，导致服务器在高负载下实现一种平缓的性能降低。

    可通过如下方式设置饱和策略：

          executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());

 

六：线程工厂

（1）每当线程池需要创建一个线程时，都是通过线程工厂方法来完成的。默认的线程工厂方法将创建一个新的、非守护的线程，并且不包含特殊的配置信息。通过指定一个线程工厂方法，可以定制线程池的配置信息。在ThreadFactory中只定义了一个方法newThread，每当线程池需要创建一个新线程时都会调用这个方法。

（2）许多情况下都需要使用定制的线程工厂方法。例如，你希望为线程池中的线程指定一个UncaughtExceptionHandler，或者实例化一个定制的Thread类用于执行调试信息的记录。你还可能希望修改线程的优先级（这通常并不是一个好主意）或者守护状态（同样，这也不是一个好主意）。或许你只是希望给线程取一个更有意义的名称，用来解释线程的转储信息和错误日志。

public interface ThreadFactory {

    /**
     * Constructs a new {@code Thread}.  Implementations may also initialize
     * priority, name, daemon status, {@code ThreadGroup}, etc.
     *
     * @param r a runnable to be executed by new thread instance
     * @return constructed thread, or {@code null} if the request to
     *         create a thread is rejected
     */
    Thread newThread(Runnable r);
}

 

通过实现该接口，可以定制自己的线程池工厂方法。

 

七：调用构造函数后再定制ThreadPoolExecutor

（1）在调用完ThreadPoolExecutor的构造函数后，仍然可以通过设置函数（setter）来修改大多数传递给它的构造函数的参数（例如线程池的基本大小、最大大小、存活时间、线程工厂以及拒绝执行处理器）。如果Executor是通过Executors中的某个（newSingleThreadExecutor除外）工厂方法创建的，那么可以将结果的类型转换为ThreadPoolExecutor以访问设置器，如下：

    ExecutorService exec = Executors.newCachedThreadPool();

       if ( exec instanceof ThreadPoolExecutor)

              ((ThreadPoolExecutor) exec).setCorePoolSize(10);

      else

              throw new AssertionError("Oops, bad assumption");

 

八：扩展ThreadPoolExecutor

（1）ThreadPoolExecutor是可扩展的，它提供了几个可以在子类化中改写的方法：beforeExecute、afterExecute和 terminated，这些方法可以用于扩展ThreadPoolExecutor的行为。

（2）在执行任务的线程中将调用beforeExecute 和 afterExecute等方法，在这些方法中还可以添加日志、计时、监视或统计信息收集的功能。无论任务是从run中正常返回还是抛出一个异常而返回，afterExecute都会被调用。（如果任何在调用完成后带有一个Error，那么就不会调用afterExecute。）如果beforeExecute抛出一个RuntimeException，那么任务将不被执行，并且afterExecute也不会被调用。

（3）在线程池完成关闭操作时调用terminated，也就是在所有任务都已经完成并且所有工作者线程也已经关闭后。terminated可以用来释放Executor在其生命周期里分配的各种资源，此外还可以执行发送通知、记录日志或者收集finalize统计信息等操作。