ThreadPoolExecutor 分析

àì夳堔傛蜴生んèń 2022-08-23 05:38 168阅读 0赞

自定义的ThreadPoolExecutor, 代码如下：

<table> 
 <tbody> 
 <tr> 
 <td> public void init() { &nbsp;&nbsp;&nbsp;&nbsp;this.service = new ThreadPoolExecutor(30, 200, 60L, TimeUnit.SECONDS, new SynchronousQueue&lt;Runnable&gt;(), &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;new CustomizableThreadFactory("ruleExecute")); } </td> 
 </tr> 
 </tbody> 
</table>

**ThreadPoolExecutor的构造函数**

<table> 
 <tbody> 
 <tr> 
 <td> public ThreadPoolExecutor(int corePoolSize, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;int maximumPoolSize, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;long keepAliveTime, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;TimeUnit unit, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;BlockingQueue&lt;Runnable&gt; workQueue, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;ThreadFactory threadFactory, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;RejectedExecutionHandler handler) </td> 
 </tr> 
 </tbody> 
</table>

各个字段含义如下：

<table> 
 <tbody> 
 <tr> 
 <td> (1)corePoolSize: 线程池中的线程个数 (2)maximumPoolSize: 线程池中的允许最大线程个数 (3)keepAliveTime &amp; unit: 当线程池中某个线程空闲时间大于该值，则回收，释放资源给OS，注意：这个只会释放多于corePoolSize的线程， 如要释放coreThread需调用：allowCoreThreadTimeOut(true) (4)workQueue：当线程池中的线程都处于运行状态，则新加入的job将会放入到workQueue中等待执行 (5)threadFactory: 创建线程的工厂类 (6)handler: 当workQueue满了，而且线程池中线程数已经大于maximumPoolSize，此时，新加入的job将不会被正常执行，转而由handler来执行，这种情况下job将不会得到正常执行。 </td> 
 </tr> 
 </tbody> 
</table>

总体而言，整个ThreadPoolExecutor是个典型的生产者-消费者模式，它的工作流程如下：

<table> 
 <tbody> 
 <tr> 
 <td> 1. 当线程池中线程数少于corePoolSize, 每收到一个job，则创建一个新的线程来执行 2. 当线程池中线程数等于corePoolSize，每收到一个job，将此job加入到workQueue 3. 当workQueue也满了，则每收到一个job，创建一个线程来执行，直到线程池中的线程数达到maximumPoolSize 4. 当workQueue满了而且线程池中线程数也达到了maximumPoolSize，则新加入的job将会不被执行，转由handler来执行 </td> 
 </tr> 
 </tbody> 
</table>

所以不同的BlockingQueue会影响ThreadPoolExecutor的执行：

<table> 
 <tbody> 
 <tr> 
 <td> 1.ArrayBlockingQueue，有界队列，内存中顺序存储，它严格按照上面流程来工作 2.LinkedBlockingQueue, 默认情况下无界队列，这种情况下maximumPoolSize将不起作用，线程池中将用于只有corePoolSize个线程在工作，job来者照收，全部放入到workQueue直到撑爆堆内存(如果线程执行不快) 3.SynchronousQueue, 这种队列有个特性，加入其中的job只有被线程执行了，才会接受下一个job，所以一般情况下需要将maximumPoolSize设置很大，这样，同一时间会创建很多线程来执行不同的job，流程如下：此时，线程池中已经有corePoolSize个线程在运行了，来个job A和job B，job A放入到SynchronousQueue，按照此队列的特性，job B放不进去(等价于workQueue满了)，按照流程3，要创建线程数来执行job B，所以，如果maximumPoolSize不维持一个比较大的数，则会进入流程4。在权衡资源的情况下，这种队列的线程池的执行效率会非常高，同时CPU也会非常的忙。 </td> 
 </tr> 
 </tbody> 
</table>

所以，按照上面逻辑，看下几种常用的ThreadPoolExecutor：

**newSingleThreadExecutor**

<table> 
 <tbody> 
 <tr> 
 <td> public static ExecutorService newSingleThreadExecutor() { &nbsp;&nbsp;&nbsp;&nbsp;return new FinalizableDelegatedExecutorService &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;(new ThreadPoolExecutor(1, 1, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;0L, TimeUnit.MILLISECONDS, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;new LinkedBlockingQueue&lt;Runnable&gt;())); } </td> 
 </tr> 
 </tbody> 
</table>

特点：同一时间，线程池中永远只有一个线程在执行，其它都在workqueue中休息，CPU利用率太低, 因为是LinkedBlockingQueue，可能会耗尽内存

**newFixedThreadPool**

<table> 
 <tbody> 
 <tr> 
 <td> public static ExecutorService newFixedThreadPool(int nThreads) { &nbsp;&nbsp;&nbsp;&nbsp;return new ThreadPoolExecutor(nThreads, nThreads, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;0L, TimeUnit.MILLISECONDS, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;new LinkedBlockingQueue&lt;Runnable&gt;()); } </td> 
 </tr> 
 </tbody> 
</table>

特点：同一时间，线程池中永远只有N个线程在执行，其它都在workqueue中休息，因为是LinkedBlockingQueue，可能会耗尽内存

**newCachedThreadPool**

<table> 
 <tbody> 
 <tr> 
 <td> public static ExecutorService newCachedThreadPool() { &nbsp;&nbsp;&nbsp;&nbsp;return new ThreadPoolExecutor(0, Integer.MAX_VALUE, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;60L, TimeUnit.SECONDS, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;new SynchronousQueue&lt;Runnable&gt;()); } </td> 
 </tr> 
 </tbody> 
</table>

特点：同一时间，线程池中最多有无限个线程在执行，至多一个等待线程，CPU利用率高，可能会耗尽内存，但是不见得效率高

其它自定义的实现BlockingQueue接口的workQueue具体情况具体分析