深入了解Java并发——《Java Concurrency in Practice》8.线程池的使用

Posted 2022-12-13 在咖啡里溺水的鱼

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

虽然Executor框架为指定和修改执行策略都提供了相当大的灵活性，但并非所有的任务都能适用所有的执行策略。有些类型的任务需要明确的指定执行策略：

依赖性任务

如果提交给线程池的任务需要依赖其他任务，那么就隐含的给执行策略带来了约束。此时必须小心的维持这些执行策略以避免产生活跃性问题。

使用线程封闭机制的任务

单线程的Executor能够对并发性做出更强的承诺。它能够确保任务不会并发执行，对象可以封闭在任务线程中，不需要同步，不是线程安全的也可以。这种情形将在任务与执行策略形成隐式的耦合——任务要求其执行所在的Executor是单线程的。如果将Executor从单线程环境改为线程池环境，那么将会失去线程安全性

响应时间敏感的任务

如果将一个运行时间较长的任务提交到单线程的Executor中，或者将多个运行时间较长的任务提交到一个只包含少量线程的线程池中，那么将降低该Executor管理的服务的响应性。

使用ThreadLocal的任务

ThreadLocal使每个线程都可以拥有某个变量的一个私有版本。然而，只要条件允许，Executor可以自由的重用这些线程。只有当线程本地值的声明周期受限于任务的生命周期时，在线程池的线程中使用ThreadLocal才有意义，而在线程池的线程中不应该使用ThreadLocal在任务之间传递值。

只有当任务都是同类型的并且相互独立时，线程池的性能才能达到最佳。

8.1.1 线程饥饿死锁

线程池中的任务需要无限期的等待一些必须由池中其他任务才能提供的资源或条件，那么除非线程池足够大，否则将发生线程饥饿死锁。

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

8.1.2 运行时间较长的任务

限定任务等待资源的时间，可以缓解执行时间较长任务造成的影响，而不要无限制的等待。在平台类库的大多数可阻塞方法中，都同时定义了限时版本和无限时版本。如果等待超时，可以把任务标识为失败，然后中止任务或者将任务重新放回队列以便随后执行。这样，无论任务的最终结果是否成功，都能确保任务总能继续执行下去，并将线程释放出来以执行一些能更快完成的任务。

8.2 设置线程池的大小

在代码中通常不会固定线程池的大小，而应该通过某种配置机制来提供，或者根据Runtime.availableProcessors来动态计算。

如果需要执行不同类别的任务，并且它们之间的行为相差很大，那么应该考虑使用多个线程池，从而使每个线程池可以根据各自的工作负载来调整。

对于计算密集型的任务，在拥有N-cpu个处理器的系统上，当线程池的大小为N-cpu+1时，通常能实现最优的利用率。对于包含I/O操作或其他阻塞操作的任务，由于线程不会一直执行，因此线程池的规模应该更大。

给出下列定义：

N_cpu=\\left(numbe \\; rof \\; CPU_s \\right)

U_cpu=target \\; CPU \\; utilization,0 \\leqslant U_cpu \\leqslant 1

\\fracWC = ratio \\; of \\; wati \\; time\\; to\\; compute \\; time

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

N_threads = N_cpu \\times U_cpu \\times \\left(1 + \\fracWC\\right)

可以通过Runtime获取CPU数目

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

CPU周期并不是唯一影响线程池大小的资源，内存、文件句柄、套接字句柄、数据库连接等都会对线程池大小的上限产生限制。

8.3 配置ThreadPoolExecutor

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

如果默认的执行策略不能满足需求，可以通过ThreadPoolExecutor的构造函数来实例化一个对象，根据自己的需求来定制。

public ThreadPoolExecutor(
    int corePoolSize,
    int maximumPoolSize,
    TimeUnit unit,
    BlockingQueue<Runnable> workQueue,
    ThreadFactory threadFactory,
    RejectedExecutionHandler handler
)

8.3.1 线程的创建与销毁

线程池的基本大小 Core Pool Size、最大大小 Maximum Pool Size、存活时间等因素共同负责线程的创建与销毁。

基本大小

线程池的目标大小，即在没有任务执行时线程池的大小，只有在工作队列满了的情况下才会创建超出这个数量的线程

最大大小

可同时活动的线程数量的上限。

存活时间

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

newFixedThreadPool将线程池的基本大小和最大大小设置为参数中指定的值，创建的线程池不会超时。
newCachedThreadPool将线程池的最大大小设置为Integer.MAX_VALUE，基本大小设置为0，超时设置为1分钟。这种线程池可以被无限扩展，需求降低时会自动收缩。

8.3.2 管理队列任务

在有限的线程池中会限制可并发执行的任务数量。

如果新请求的到达速率超过了线程池的处理速率，那么新到来的请求会在一个由Executor管理的Runnable队列中等待。

尽管队列有助于缓解任务突增问题，但如果任务持续高速到来，最终还是会一致请求的到达率以避免耗尽内存。内存耗尽之前，响应性能也随着任务队列的增长而变得糟糕。

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

无界队列

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

有界队列

更稳妥的资源管理策略是使用有界队列，如ArrayBlockingQueue、有界的LinkedBlockingQueue、PriorityBlockingQueue。有界队列有助于避免资源耗尽的情况发生。在使用有界队列时，队列的大小与线程池的大小必须一起调节。线程池较小而队列较大，有助于减少内存使用量和CPU使用率，同时还可以减少上下文切换，但代价是可能会限制吞吐量。

SynchronouseQueue

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

执行顺序

使用LinkedBlockingQueue或ArrayBlockingQueue这样的FIFO队列时，任务的执行顺序与他们的到达顺序相同。如果想进一步控制任务执行顺序，可以使用PriorityBlockingQueue，这个队列将根据优先级来安排任务。任务的优先级是通过自然顺序或Comparator（如果任务实现了Comparable）定义的。

对于Executor，newCachedThreadPool工厂方法是一种很好的默认选择，它能比固定大小的线程池更好的排队性能。当需要限制当前任务的数量以满足资源管理需求时，那么可以选择固定大小的线程池，就像在接受网络客户请求的服务器应用程序中，如果不进行限制，那么很容易发生过载问题。

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

8.3.3 饱和策略

有界队列被填满时，饱和策略开始发挥作用。可以通过setRejectedExecutionHandler来修改ThreadPoolExecutor的饱和策略。JDK提供了集中不同的RejectedExecutionHandler实现，对应与不同的饱和策略。

AbortPolicy

中止 策略 是默认的饱和策略，该策略抛出非受检异常RejectedExecutionException。调用者可以捕获这个异常，根据需求编写自己的处理代码。当新提交的任务无法保存到队列中等待执行时， 抛弃 Discard 策略会静默的抛弃该任务，抛弃最旧 Discard-Oldest 策略会抛弃下一个将被执行的任务，然后尝试重新提交任务。对于优先队列，Discard-Oldest策略将抛弃优先级最高的任务，最好不要将Discard-Oldest饱和策略与优先级队列放在一起使用

CallerRunsPolicy

调用者运行 策略 实现了一种调节机制，该策略不会抛弃任务，也不会抛弃任务，将某些任务回退到调用者，从而降低新任务的流量。它不会再线程池的某个线程中执行新提交的任务，而是在一个调用了execute的线程中执行该任务。

创建一个固定大小的线程池，并采用有界队列、调用者运行饱和策略

ThreadPoolExecutor executor = new ThreadPoolExecutor(N_THREADS, N_THREADS, 0L, TimeUnit.MILLISECONDES, new LinkedBlockingQueue<Runnable>(CAPACITY));
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());

工作队列被填满后，没有预定义的饱和策略来阻塞execute。可以通过使用Semaphore 信号量 来限制任务的到达率，就可以实现这个功能。

8.3.4 线程工厂

线程池创建线程都是通过线程工厂方法完成的。默认的线程工厂方法将创建一个新的、非守护的线程，不包含特殊的配置信息。通过指定一个线程工厂方法，可以定制线程池的配置信息。

如果在应用程序中需要利用安全策略来控制对某些特殊代码库的访问权限，那么可以通过Executor中的privilegedThreadFactory工厂来定制自己的线程工厂。通过这种方式创建出来的线程，将与创建privilegedThreadFactory的线程拥有相同的访问权限、AccessControlContext和contextClassLoader。

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

调用完ThreadPoolExecutor的构造函数后，仍然可以通过setter方法来修改大多数传递给它构造器的方法。如果是通过Executors中的除了newSingleThreadExceutor外某个工厂方法创建的，那么可以将结果的类型转换为ThreadPollExecutor以访问设置器。

在Executors中包含一个unconfigurableExecutorService工厂方法，该方法对一个现有的ExecutorService进行包装，使其只暴露出ExecutorService的方法，因此不能对它进行配置。可以使用这项技术以防止执行策略被修改。

8.4 扩展ThreadPoolExecutor

ThreadPoolExecutor提供了几个可以再子类化中改写的方法：beforeExecute、afterExecute、terminated，这些方法可以用于扩展ThreadPoolExecutor的行为。可以利用这几个方法进行日志、计时、监控、统计、资源回收等操作。

8.5 递归算法的并行化

如果循环中的迭代操作都是独立的，并且不需要等待所有的迭代器都完成再继续执行，就可以使用Executor将串行循环转化为并行循环。

void processSequentially(List<Element> elements) 
    for(Element e : elements) 
        process(e);
    

void processInParallel(Executor exec, List<Element> elements) 
    for(final Element e : elements) 
        exec.execute(new Runnable() 
            public void run() 
                process(e);
                
        );