线程池与线程队列分析-优

川长思鸟来 2022-07-17 00:45 222阅读 0赞

·  线程池是对象池的一个有用的例子，它能够节省在创建它们时候的资源开销。并且线程池对系统中的线程数量也起到了很好的限制作用。

·  线程池中的线程数量必须仔细的设置，否则冒然增加线程数量只会带来性能的下降。

·  在定制ThreadPoolExecutor时，遵循KISS原则，通常情况下会提供最好的性能。

线程池类为java.util.concurrent.ThreadPollExecutor,常用构造方法为：ThreadPoolExecutor(

int corePoolSize,

int maxnumPollSize,

long keepAliveTime,

TimeUnit unit,

BlockingQueue<Runnable> workQueue,

RejectedExecutionHandler handler)，各个属性介绍：

corePoolSize: 线程池维护线程的最少数量

maxnumPoolSize:线程池维护线程的最大数量

keepAliveTime:线程池维护线程所允许的空闲时间

unit:线程池维护线程所允许的空闲时间的单位

workQueue:线程池所使用的缓冲队列

handler：线程池对拒绝任务的处理策略

一个任务通过execute(Runnable)方法被添加到线程池，任务就是一个Runnable类型的对象，任务的执行方法就是Runnable类型对象的run()方法。

当一个任务通过execute(Runnable)方法想添加到线程池时：

1.      如果此时线程池中数量小于corePoolSize,即使线程池中的线程都处于空闲状态，也要创建新的线程来处理被添加的任务。

2.      如果此时线程池中的数量等于corePoolSize,但是缓冲队列workQueue未满，那么任务放入缓冲队列

3.      如果此时线程池中的数量大于corePoolSize，缓冲队列workQueue满，并且线程池中的数量小于maximumPoolSize，建新的线程来处理被添加的任务。

4.      如果此时线程池中的数量大于corePoolSize，缓冲队列workQueue 满，并且线程池中的数量等于maximumPoolSize，那么通过 handler所指定的策略来处理此任务。也就是：处理任务的优先级为：核心线程 corePoolSize、任务队列workQueue、最大线程maximumPoolSize，如果三者都满了，使用handler处理被拒绝的任务

5.      当线程池中的线程数量大于 corePoolSize时，如果某线程空闲时间超过keepAliveTime，线程将被终止。这样，线程池可以动态的调整池中的线程数。

unit可选的参数为java.util.concurrent.TimeUnit中的几个静态属性：

NANOSECONDS、

MICROSECONDS、

MILLISECONDS、

SECONDS。

workQueue常用的是：java.util.concurrent.ArrayBlockingQueue

handler有四个选择：

ThreadPoolExecutor.AbortPolicy()

抛出java.util.concurrent.RejectedExecutionException异常

ThreadPoolExecutor.CallerRunsPolicy()

当抛出RejectedExecutionException异常时，会调用**rejectedExecution**方法

(如果主线程没有关闭，则主线程调用run方法,源码如下

publicvoid rejectedExecution(Runnable r, ThreadPoolExecutor e) \{  
            if (!e.isShutdown()) \{  
                r.run();  
            \}  
        \}

)

ThreadPoolExecutor.DiscardOldestPolicy()

抛弃旧的任务

ThreadPoolExecutor.DiscardPolicy()

抛弃当前的任务

线程池可以解决两个不同问题：由于减少了每个任务调用的开销，它们通常可以在执行大 量异步任务时提供增强的性能，并且还可以提供绑定和管理资源（包括执行集合任务时使用的线程）的方法。每个ThreadPoolExecutor 还维护 着一些基本的统计数据，如完成的任务数。

**排队有三种通用策略：**

直接提交。工作队列的默认选项是 SynchronousQueue，它将任务直接提 交给线程而不保持它们。在此，如果不存在可用于立即运行任务的线程，则试图把任务加入队列将失败，因此会构造一个新的线程。此策略可以避免在处理可能具有 内部依赖性的请求集合时出现锁定。直接提交通常要求无界 maximumPoolSizes 以避免拒绝新提交的任务。当命令以超过队列所能处理的平均数 连续到达时，此策略允许无界线程具有增长的可能性。

无界队列。使用无界队列（例如，不具有预定义容量 的 LinkedBlockingQueue）将导致在所有 corePoolSize 线程都忙的情况下将新任务加入队列。这样，创建的线程就不会超 过 corePoolSize。（因此，maximumPoolSize 的值也就无效了。）当每个任务完全独立于其他任务，即任务执行互不影响时，适合 于使用无界队列；例如，在 Web 页服务器中。这种排队可用于处理瞬态突发请求，当命令以超过队列所能处理的平均数连续到达时，此策略允许无界线程具有 增长的可能性。

有界队列。当使用有限的 maximumPoolSizes 时，有界队列 （如 ArrayBlockingQueue）有助于防止资源耗尽，但是可能较难调整和控制。队列大小和最大池大小可能需要相互折衷：使用大型队列和小型 池可以最大限度地降低CPU 使用率、操作系统资源和上下文切换开销，但是可能导致人工降低吞吐量。如果任务频繁阻塞（例如，如果它们是 I/O 边 界），则系统可能为超过您许可的更多线程安排时间。使用小型队列通常要求较大的池大小，CPU 使用率较高，但是可能遇到不可接受的调度开销，这样也会降 低吞吐量。

**挂钩方法**

此类提供 protected 可重写 的 beforeExecute(java.lang.Thread,java.lang.Runnable) 和 afterExecute(java.lang.Runnable,java.lang.Throwable) 方法，这两种方法分别在执行每个任务之前和之后调用。它们可用于操纵执行环境；例如，重新初始化 ThreadLocal、搜集统计信息或添加日志条目。此外，还可以重写方法 terminated() 来执行 Executor 完全终止后需要完成 的所有特殊处理。

如果挂钩或回调方法抛出异常，则内部辅助线程将依次失败并突然终止。

**队列维护**

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

强烈建议程序员使用较为方便的 Executors 工厂方 法 Executors.newCachedThreadPool()（无界线程池，可以进行自动线程回收）、Executors.newFixedThreadPool(int)（固定大小线程池） 和 Executors.newSingleThreadExecutor()（单个后台线程），它们均为大多数使用场景预定义了设置。

newFixedThreadPool和newSingleThreadExectuor默认使用的是一个无限的LinkedBlockingQueue。如果所有的工作者线程都处于忙碌状态，任务会在队列中等候。如果任务持续快速到达，超过了它们被执行的速度，队 列也会无限制地增加。稳妥的策略是使用有限队列，比如ArrayBlockingQueue或有限的LinkedBlockingQueue以及 PriorityBlockingQueue。newCachedThreadPool使用SynchronousQueue，完全绕开队列，直接将任务由生产者交给工作者线程。非常重要的一点，可以使用PriorityBlockingQueue通过优先级安排任务，这样几个子任务严格按照顺序执行的时候，可以用这个来实现哈。

**newFixedThreadPool()/newSingleThreadPool()**

创建线程数固定大小的线程池。 由于使用了LinkedBlockingQueue所以maximumPoolSize 没用(除非用ArrayBlockingQueue或者指定linkedBlockingQueue的长度)，当corePoolSize满了之后就加入到LinkedBlockingQueue队列中。每当某个线程执行完成之后就从LinkedBlockingQueue队列中取一个。所以这个是创建固定大小的线程池。

无界限的队列可以是诸如LinkedBlockingQueue这种类型，在这种情况下，任何被提交的任务都不会被拒绝。但是线程池会忽略最大线程数这一参数，意味着线程池的最大线程数就变成了设置的最小线程数。所以在使用这种队列时，通常会将最大线程数设置的和最小线程数相等。这就相当于使用了一个固定 了线程数量的线程池。

**newCachedThreadPool()**

创建可缓冲的线程池。没有大小限制。由于corePoolSize为0所以任务会放入SynchronousQueue队列中，SynchronousQueue只能存放大小为1，所以会立刻新起线程，由于maxumumPoolSize为Integer.MAX\_VALUE所以可以认为大小为2147483647。受内存大小限制。任务队列是 SynchronousQueue 这个队列的特点是，它并不能放置任何任务在其队列中，当有任务被提交时，使用SynchronousQueue的线程池会立即为该任务创建一个线程(如果 线程数量没有达到最大时，如果达到了最大，那么该任务会被拒绝)。这种队列适合于当任务数量较小时采用。也就是说，在使用这种队列时，未被执行的任务没有 一个容器来暂时储存。

**关于BlockingQueue补充：**

BlockingQueue也是java.util.concurrent下的主要用来控制线程同步的工具。

BlockingQueue有四个具体的实现类,根据不同需求,选择不同的实现类  
1、ArrayBlockingQueue：一个由数组支持的有界阻塞队列，规定大小的BlockingQueue,其构造函数必须带一个int参数来指明其大小.其所含的对象是以FIFO(先入先出)顺序排序的。

2、LinkedBlockingQueue：大小不定的BlockingQueue,若其构造函数带一个规定大小的参数,生成的BlockingQueue有大小限制,若不带大小参数,所生成的 BlockingQueue的大小由Integer.MAX\_VALUE来决定.其所含的对象是以FIFO(先入先出)顺序排序的。

3、PriorityBlockingQueue：类似于LinkedBlockQueue,但其所含对象的排序不是FIFO,而是依据对象的自然排序顺序或者是构造函数的Comparator决定的顺序。

4、SynchronousQueue：特殊的BlockingQueue,对其的操作必须是放和取交替完成的。

LinkedBlockingQueue 可以指定容量，也可以不指定，不指定的话，默认最大是Integer.MAX\_VALUE,其中主要用到put和take方法，put方法在队列满的时候 会阻塞直到有队列成员被消费，take方法在队列空的时候会阻塞，直到有队列成员被放进来

在[Java][]7中引入了一种新的线程池：ForkJoinPool。

它同ThreadPoolExecutor一样，也实现了Executor和ExecutorService接口。它使用了一个无限队列来保存需要执行的 任务，而线程的数量则是通过构造函数传入，如果没有向构造函数中传入希望的线程数量，那么当前计算机可用的CPU数量会被设置为线程数量作为默认值。

ForkJoinPool主要用来使用分治法(Divide-and-ConquerAlgorithm)来解决问题。典型的应用比如快速排序算法。这里的要点在于，ForkJoinPool需要使用相对少的线程来处理大量的任务。比如要对1000万个数据进行排序，那么会将这个任务分割成两个500万的排序任务和一个针对这两组500万数据的合并任务。以此类推，对于500万的数据也会 做出同样的分割处理，到最后会设置一个阈值来规定当数据规模到多少时，停止这样的分割处理。比如，当元素的数量小于10时，会停止分割，转而使用插入排序 对它们进行排序。

那么到最后，所有的任务加起来会有大概2000000+个。问题的关键在于，对于一个任务而言，只有当它所有的子任务完成之后，它才能够被执行。

所以当使用ThreadPoolExecutor时，使用分治法会存在问题，因为**ThreadPoolExecutor中的线程无法像任务队列中再添加一个任务并且在等待该任务完成之后再继续执行。而使用ForkJoinPool时，就能够让其中的线程创建新的任务，并挂起当前的任务，此时线程就能够从队列中选择子任务执行**

**ForkJoinPool的另外一个特性是它能够实现工作窃取(Work Stealing)**，在该线程池的每个线程中会维护一个队列来存放需要被执行的任务。当线程自身队列中的任务都执行完毕后，它会从别的线程中拿到未被执行的任务并帮助它执行。

可以通过以下的代码来测试ForkJoinPool的Work Stealing特性：

for (int i = first; i <= last; i++) \{

if (d\[i\] <0.5) \{

subCount++;

for (int j = 0;j < d.length - i; j++) \{

for (int k= 0; k < 100; k++) \{

dummy =j \* k + i; // dummy is volatile, so multiple writes occur

d\[i\] =dummy;

因为里层的循环次数(j)是依赖于外层的i的值的，所以这段代码的执行时间依赖于i的值。当i = 0时，执行时间最长，而i= last时执行时间最短。也就意味着任务的工作量是不一样的，当i的值较小时，任务的工作量大，随着i逐渐增加，任务的工作量变小。因此这是一个典型的任 务负载不均衡的场景。

这时，选择ThreadPoolExecutor就不合适了，因为它其中的线程并不会关注每个任务之间任务量的差异。当执行任务量最小的任务的线程执行完毕后，它就会处于空闲的状态(Idle)，等待任务量最大的任务执行完毕。

而ForkJoinPool的情况就不同了，即使任务的工作量有差别，当某个线程在执行工作量大的任务时，其他的空闲线程会帮助它完成剩下的任务。因此，提高了线程的利用率，从而提高了整体性能。

**自动并行化(Automatic Parallelization)**

在Java8中，引入了自动并行化的概念。它能够让一部分Java代码自动地以并行的方式执行，前提是使用了ForkJoinPool。

Java 8为ForkJoinPool添加了一个通用线程池，这个线程池用来处理那些没有被显式提交到任何线程池的任务。它是ForkJoinPool类型上的一个静态元素，它拥有的默认线程数量等于运行计算机上的处理器数量。

当调用Arrays类上添加的新方法时，自动并行化就会发生。比如用来排序一个数组的并行快速排序，用来对一个数组中的元素进行并行遍历。自动并行化也被运用在Java 8新添加的Stream API中。

比如下面的代码用来遍历列表中的元素并执行需要的计算：

Stream stream = arrayList.parallelStream();

stream.forEach(a -> \{

String symbol =StockPriceUtils.makeSymbol(a);

StockPriceHistory sph = new StockPriceHistoryImpl(symbol, startDate,endDate, entityManager);

\});

对于列表中的元素的计算都会以并行的方式执行。forEach方法会为每个元素的计算操作创建一个任务，该任务会被前文中提到的ForkJoinPool中的通用线程池处理。以上的并行计算逻辑当然也可以使用ThreadPoolExecutor完成，但是就代码的可读性和代码量而言，使用 ForkJoinPool明显更胜一筹。

对于ForkJoinPool通用线程池的线程数量，通常使用默认值就可以了，即运行时计算机的处理器数量。如果需要调整线程数量，可以通过设置[系统][Link 1]属性：-Djava.util.concurrent.ForkJoinPool.common.parallelism=N

下面的一组数据用来比较使用ThreadPoolExecutor和ForkJoinPool中的通用线程池来完成上面简单计算时的性能：

<table> 
 <thead> 
 <tr> 
 <td> 线程数 </td> 
 <td> ThreadPoolExecutor(秒) </td> 
 <td> ForkJoinPool Common Pool(秒) </td> 
 </tr> 
 </thead> 
 <tbody> 
 <tr> 
 <td> 1 </td> 
 <td> 255.6 </td> 
 <td> 135.4 </td> 
 </tr> 
 <tr> 
 <td> 2 </td> 
 <td> 134.8 </td> 
 <td> 110.2 </td> 
 </tr> 
 <tr> 
 <td> 4 </td> 
 <td> 77.0 </td> 
 <td> 96.5 </td> 
 </tr> 
 <tr> 
 <td> 8 </td> 
 <td> 81.7 </td> 
 <td> 84.0 </td> 
 </tr> 
 <tr> 
 <td> 16 </td> 
 <td> 85.6 </td> 
 <td> 84.6 </td> 
 </tr> 
 </tbody> 
</table>

注意到当线程数为1，2，4时，性能差异的比较明显。线程数为1的ForkJoinPool通用线程池和线程数为2的ThreadPoolExecutor的性能十分接近。

出现这种现象的原因是，forEach方法用了一些小把戏。它会将执行forEach本身的线程也作为线程池中的一个工作线程。因此，即使将 ForkJoinPool的通用线程池的线程数量设置为1，实际上也会有2个工作线程。因此在使用forEach的时候，线程数为1的 ForkJoinPool通用线程池和线程数为2的ThreadPoolExecutor是等价的。

所以当ForkJoinPool通用线程池实际需要4个工作线程时，可以将它设置成3，那么在运行时可用的工作线程就是4了

[Java]: http://www.2cto.com/kf/ware/Java/
[Link 1]: http://www.2cto.com/os/