并发编程7：线程池的使用

目录

1、在任务与执行策略之间的隐性耦合

1.1 线程饥饿死锁

1.2 运行时间较长的任务

2、设置线程池的大小

3、配置 ThreadPoolExecutor

3.1 线程的创建与销毁

3.2 管理队列任务

3.3 饱和策略

3.4 线程工厂

3.5 在调用构造函数后再定制 ThreadPoolExecutor

4、扩展 ThreadPoolExecutor

---

1、在任务与执行策略之间的隐性耦合

        只有当任务都是同类型的并且相互独立时，线程池的性能才能达到最佳。如果将运行时间较长的与运行时间较短的任务混合在一起，那么除非线程池很大，否则将可能造成“拥塞”。如果提交的任务依赖于其他任务，那么除非线程池无限大，否则将可能造成死锁。

1.1 线程饥饿死锁

        在线程池中，如果任务依赖于其他任务，那么可能产生死锁。只要线池中的任务需要无限期地等待一些必须由池中其他任务才能提供的资源或条件，例如某个任务等待另一个任务的返回值或执行结果，那么除非线程池足够大，否则将发生线程饥饿死锁。//有等待就有可能出现死锁

        每当提交了一个有依赖性的 Executor 任务时，要清楚地知道可能会出现线程“饥饿”死锁，因此需要在代码或配置 Executor 的配置文件中记录线程池的大小限制或配置限制。

1.2 运行时间较长的任务

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

        有一项技术可以缓解执行时间较长任务造成的影响，即限定任务等待资源的时间，而不要无限制地等待。在平台类库的大多数可阻塞方法中，都同时定义了限时版本和无限时版本，例如 Thread.join、BlockingQueue.put、CountDownLatch.await 以及 Selector.select 等。如果等待超时，那么可以把任务标识为失败，然后中止任务或者将任务重新放回队列以便随后执行。这样，无论任务的最终结果是否成功，这种办法都能确保任务总能继续执行下去，并将线程释放出来以执行一些能更快完成的任务。如果在线程池中总是充满了被阻塞的任务，那么也可能表明线程池的规模过小。//如果限定的时间小于长任务的正常执行时间，那么长任务将一直无法完成

2、设置线程池的大小

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

        对于计算密集型的任务，在拥有 N(cpu) 个处理器的系统上，当线池的大小为 N+I 时，通常能实现最优的利用率。即使当计算密集型的线程偶尔由于页缺失故障或者其他原因而暂停时，这个“额外”的线程也能确保 CPU 的时钟周期不会被浪费。//CPU占用率高就需要根据CPU数量来分配，确保CPU能一直忙碌

        对于包 I/O 操作或者其他阻塞操作的任务，由于线程并不会一直执行，因此线程池的规模应该更大。

数据参考：

CPU密集型：操作内存处理的业务，一般线程数设置为：CPU核数 + 1 或者 CPU核数 * 2。核数为 4 的话，一般设置 5 或 8

I/O密集型：文件操作，网络操作，数据库操作，一般线程设置为：CPU核数 / (1-0.9)，核数为 4 的话，一般设置为 40

        CPU 周期并不是唯一影响线程池大小的资源，还包括内存、文件句柄、套接字句柄和数据库连接等。

        当任务需要某种通过资源池来管理的资源时，例如数据库连接，那么线程池和资源池的大小将会相互影响。如果每个任务都需要一个数据库连接，那么连接池的大小就限制了线程池的大小。同样，当线程池中的任务是数据库连接的唯一使用者时，那么线程池的大小又将限制连接池的大小。//如果使用到数据库，那么数据库连接池大小会现在线程连接池的大小

3、配置 ThreadPoolExecutor

        ThreadPoolExecutor 为一些 Executor 提供了基本的实现，这些 Executor 是由 Executors 中的 newCachedThreadPool、newFixedThreadPool和 newScheduledThreadExecutor 等工厂方法返回的。ThreadPoolExecutor 是一个灵活的、稳定的线程池，允许进行各种定制。

    public ThreadPoolExecutor(int corePoolSize,     //核心线程
                              int maximumPoolSize,  //最大线程
                              long keepAliveTime,   //存活时间
                              TimeUnit unit,        //时间单位
                              BlockingQueue<Runnable> workQueue, //阻塞队列
                              ThreadFactory threadFactory,       //线程工厂
                              RejectedExecutionHandler handler) {//拒绝策略
        if (corePoolSize < 0 ||
            maximumPoolSize <= 0 ||
            maximumPoolSize < corePoolSize ||
            keepAliveTime < 0)
            throw new IllegalArgumentException();
        if (workQueue == null || threadFactory == null || handler == null)
            throw new NullPointerException();
        this.acc = System.getSecurityManager() == null ?
                null :
                AccessController.getContext();
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }

3.1 线程的创建与销毁

        线程池的基本大小(Core Pool Size)、最大大小(Maximum Pool Size) 以及存活时间等因素共同负责线程的创建与销毁。基本大小也就是线程池的目标大小，即在没有任务执行时线程池的大小，并且只有在工作队列满了的情况下才会创建超出这个数量的线程。线程池的最大大小表示可同时活动的线程数量的上限。如果某个线程的空闲时间超过了存活时间，那么将被标记为可回收的，并且当线程池的当前大小超过了基本大小时，这个线程将被终止。//基于这个工作队列满，可以回想下工厂招聘的那个示例，适当的订单积压并不需要立即招聘，只有当订单积压到一定程度才回去招募临时工

3.2 管理队列任务

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

        当使用有界队列进行资源管理时，例如 ArrayBlockingQueue、有界的
LinkedBlockingQueue、PriorityBlockingQueue。有界队列有助于避免资源耗尽的情况发生，
但它又带来了新的问题：当队列填满后，新的任务该怎么办？//所以，使用有界队列时，必须配置对应的拒绝策略 -> 有界队列+拒绝策略

        对于非常大的或者无界的线程池，可以通过使用 SynchronousQueue 来避免任务排队，以及直接将任务从生产者移交给工作者线程。SynchronousQueue 不是一个真正的队列，而是一种在线程之间进行移交的机制。要将一个元素放入 SynchronousQueue 中，必须有另一个线程正在等待接受这个元素。如果没有线程正在等待，并且线程池的当前大小小于最大值，那么ThreadPoolExecutor 将创建一个新的线程，否则根据饱和策略，这个任务将被拒绝。使用直接移交将更高效，因为任务会直接移交给执行它的线程，而不是被首先放在队列中，然后由工作者线程从队列中提取该任务。只有当线程池是无界的或者可以拒绝的任务时，SnchronousOueue 才有实际价值。在 newCachedThreadPool 工厂方法中就使用了 SynchronousQueue。

        只有当任务相互独立时，为线程池或工作队列设置界限才是合理的。如果任务之间存在依赖性，那么有界的线程池或队列就可能导致线程“饥饿”死锁问题。此时应该使用无界的线程池，例如 newCachedThreadPool。//一般都会尽量避免这种情况，脱离控制的程序往往无法把控

3.3 饱和策略

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

        //可以自定义拒绝策略 -> 阻塞提交，这里记录日志应该是记录失败的次数，如果记录请求的数据其实没有多大意义，因为此请求并没有真正执行，数据也并无作用

3.4 线程工厂

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

        在一些情况下可能需要使用定制的线程工厂方法。例如，希望为线程池中的线程指定一个UncaughtExceptionHandler，或者实例化一个定制的 Thread 类用于执行调试信息的记录。还可能希望修改线程的优先级或者守护状态(正常情况下，别这么做)。或许只是希望给线程取一个更有意义的名称，用来解释线程的转储信息和错误日志。//一般线程工厂方法都是不需要去改动的，实际情况下也不建议去改动

3.5 在调用构造函数后再定制 ThreadPoolExecutor

        在调用完 ThreadPoolExecutor 的构造函数后，仍然可以通过设置函数 Setter 来修改大多数传递给它的构造函数的参数。例如，线程池的基本大小、最大大小、存活时间、线程工厂以及拒绝执行处理器 Reiected Execution Handler。如果 Executor 是通过 Executors 中的某个 (newSingleThreadExecutor除外)工厂方法创建的，那么可以将结果的类型转换为ThreadPoolExecutor 以访问设置器。//慎用，目前该功能看起来比较鸡肋

4、扩展 ThreadPoolExecutor

        ThreadPoolExecutor 是可扩展的，它提供了几个可以在子类化中改写的方法： beforeExecute、afterExecute 和 terminated，这些方法可以用于扩展 ThreadPoolExecutor 的行为。在执行任务的线程中将调用 beforeExecute 和 afterExecute 等方法，在这些方法中还可以添加日志、计时、监视或统计信息收集的功能。无论任务是从 run 中正常返回，还是抛出一个异常而返回，afterExecute 都会被调用，但是，如果任务在完成后带有一个 Error，那么就不会调用afterExecute。如果 beforeExecute 抛出一个 RuntimeException，那么任务将不被执行，并且 afterExecute 不会被调用。//任务调用前和调用后执行操作

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

        //通过这项扩展，可以类比常用的框架，这些框架都能在一项功能执行时支持添加额外的操作，给框架带来灵活性，比如 Spring

        给线程池添加统计信息示例程序如下所示：

import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.logging.Logger;
 
public class TimingThreadPool extends ThreadPoolExecutor {
 
    public TimingThreadPool() {
        super(1, 1, 0L, TimeUnit.SECONDS, new LinkedBlockingDeque<>(10));
    }
 
    private final ThreadLocal startTime = new ThreadLocal<>();
    private final Logger            log       = Logger.getLogger("TimingThreadPool");
    private final AtomicLong        numTasks  = new AtomicLong();
    private final AtomicLong        totalTime = new AtomicLong();
 
    /**
     * 执行前
     */
    protected void beforeExecute(Thread t, Runnable r) {
        super.beforeExecute(t, r);
        log.info(String.format("[beforeExecute]:Thread %s: start %s", t, r));
        startTime.set(System.nanoTime());
    }
 
    /**
     * 执行后
     */
    protected void afterExecute(Runnable r, Throwable t) {
        try {
            long endTime = System.nanoTime();
            long taskTime = endTime - startTime.get();
            numTasks.incrementAndGet();
            totalTime.addAndGet(taskTime);
            log.info(String.format("[afterExecute]:Thread %s: end %s, time=%dns", t, r, taskTime));
        } finally {
            super.afterExecute(r, t);
        }
    }
 
    /**
     * 执行终止
     */
    protected void terminated() {
        try {
            log.info(String.format("[terminated]:Terminated: avg time=%dns", totalTime.get() / numTasks.get()));
        } finally {
            super.terminated();
        }
    }
 
    public static void main(String[] args) {
        TimingThreadPool threadPoolExecutor = new TimingThreadPool();
        for (int i = 0; i < 2; i++) {
            threadPoolExecutor.execute(() -> {
                System.out.println("[START]:开始执行任务");
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("[END]:执行任务结束");
            });
        }
        threadPoolExecutor.shutdown();
        try {
            //主线程阻塞，相当于join()
            threadPoolExecutor.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
            System.out.println("[STOP]:线程池关闭");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

        上述示例重新定义了一个新的线程池，重写了 ThreadPoolExecutor 的三个扩展方法。

        小结

        对于并发执行的任务，Executor 框架是一种强大且灵活的框架。它提供了大量可调节的选项，例如创建线程和关闭线程的策略，处理队列任务的策略，处理过多任务的策略，并且提供了几个钩子方法来扩展它的行为。然而，与大多数能强大的框架一样，其中有些设置参数并不能很好地工作，某些类型的任务需要特定的执行策略，而一些参数组合则可能产生奇怪的结果。

        至此，全文结束。

        //线程池的使用并不复杂，它降低了并发编程的难度，提供了线程的管理，是一个非常棒的工具，此外编写的每一个并发程序都需要仔细去思考，防止意外的错误