一. ThreadPoolExecutor线程池使用

我就是我 2024-03-30 17:18 81阅读 0赞

### 前言 ###

线程池是`Java`中使用较多的并发框架，合理使用线程池，可以：**降低资源消耗**，**提高响应速度**，**提高线程的可管理性**。本篇文章为`《Java并发编程的艺术》`第9章的学习笔记，根据原文作者的编写思路，依次对**线程池的原理**，**线程池的创建**，**线程池执行任务**和**关闭线程池**进行了学习和总结。

### 正文 ###

#### 一. 线程池的原理 ####

当一个任务提交到线程池`ThreadPoolExecutor`时，该任务的执行如下图所示。

![在这里插入图片描述][71fc607543f6407f84f1ebc09b554274.jpeg_pic_center]

1.  如果当前运行的线程数小于**corePoolSzie**（核心线程数），则创建新线程来执行任务（需要获取全局锁）；
2.  如果当前运行的线程数等于或大于**corePoolSzie**，则将任务加入`BlockingQueue`（任务阻塞队列）；
3.  如果`BlockingQueue`已满，则创建新的线程来执行任务（需要获取全局锁）；
4.  如果创建新线程会使当前线程数大于**maximumPoolSize**（最大线程数），则拒绝任务并调用`RejectedExecutionHandler`的`rejectedExecution()`方法。

由于`ThreadPoolExecutor`存储工作线程使用的集合是`HashSet`，因此执行上述步骤1和步骤3时需要获取全局锁来保证线程安全，而获取全局锁会导致线程池性能瓶颈，因此通常情况下，线程池完成预热后（当前线程数大于等于**corePoolSize**），线程池的`execute()`方法都是执行步骤2。

#### 二. 线程池的创建 ####

通过`ThreadPoolExecutor`能够创建一个线程池，`ThreadPoolExecutor`的构造函数签名如下。

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

通过`ThreadPoolExecutor`创建线程池时，需要指定线程池的**核心线程数**，**最大线程数**，**线程保活时间**，**线程保活时间单位**和**任务阻塞队列**，并按需指定**线程工厂**和**饱和拒绝策略**，如果不指定**线程工厂**和**饱和拒绝策略**，则`ThreadPoolExecutor`会使用默认的**线程工厂**和**饱和拒绝策略**。下面分别介绍这些参数的含义。

<table> 
 <thead> 
 <tr> 
 <th>参数</th> 
 <th>含义</th> 
 </tr> 
 </thead> 
 <tbody> 
 <tr> 
 <td>corePoolSize</td> 
 <td>核心线程数，即线程池的基本大小。当一个任务被提交到线程池时，如果线程池的线程数小于corePoolSize，那么无论其余线程是否空闲，也需创建一个新线程来执行任务。</td> 
 </tr> 
 <tr> 
 <td>maximumPoolSize</td> 
 <td>最大线程数。当线程池中线程数大于等于corePoolSize时，新提交的任务会加入任务阻塞队列，但是如果任务阻塞队列已满且线程数小于maximumPoolSize，此时会继续创建新的线程来执行任务。该参数规定了线程池允许创建的最大线程数</td> 
 </tr> 
 <tr> 
 <td>keepAliveTime</td> 
 <td>线程保活时间。当线程池的线程数大于核心线程数时，多余的空闲线程会最大存活keepAliveTime的时间，如果超过这个时间且空闲线程还没有获取到任务来执行，则该空闲线程会被回收掉。</td> 
 </tr> 
 <tr> 
 <td>unit</td> 
 <td>线程保活时间单位。通过<code>TimeUnit</code>指定线程保活时间的时间单位，可选单位有DAYS（天），HOURS（时），MINUTES（分），SECONDS（秒），MILLISECONDS（毫秒），MICROSECONDS（微秒）和NANOSECONDS（纳秒），但无论指定什么时间单位，<code>ThreadPoolExecutor</code>统一会将其转换为NANOSECONDS。</td> 
 </tr> 
 <tr> 
 <td>workQueue</td> 
 <td>任务阻塞队列。线程池的线程数大于等于corePoolSize时，新提交的任务会添加到workQueue中，所有线程执行完上一个任务后，会循环从workQueue中获取任务来执行。</td> 
 </tr> 
 <tr> 
 <td>threadFactory</td> 
 <td>创建线程的工厂。可以通过线程工厂给每个创建出来的线程设置更有意义的名字。</td> 
 </tr> 
 <tr> 
 <td>handler</td> 
 <td>饱和拒绝策略。如果任务阻塞队列已满且线程池中的线程数等于maximumPoolSize，说明线程池此时处于饱和状态，应该执行一种拒绝策略来处理新提交的任务。</td> 
 </tr> 
 </tbody> 
</table>

#### 三. 线程池执行任务 ####

线程池使用两个方法执行任务，分别为`execute()`和`submit()`。`execute()`方法用于执行不需要返回值的任务，`submit()`方法用于执行需要返回值的任务。`execute()`是接口`Executor`定义的方法，`submit()`是接口`ExecutorService`定义的方法，相关类图如下所示。

![在这里插入图片描述][a96ebb4b63b646579dbfaf11e48e4b40.jpeg_pic_center]

`ThreadPoolExecutor`对`execute()`的实现如下。

public void execute(Runnable command) {
 
 	if (command == null)
 		throw new NullPointerException();
 	int c = ctl.get();
 	if (workerCountOf(c) < corePoolSize) {
 
 		if (addWorker(command, true))
 			return;
 		c = ctl.get();
 	}
 	if (isRunning(c) && workQueue.offer(command)) {
 
 		int recheck = ctl.get();
 		if (! isRunning(recheck) && remove(command))
 			reject(command);
 		else if (workerCountOf(recheck) == 0)
 			addWorker(null, false);
 	}
 	else if (!addWorker(command, false))
 		reject(command);
 }

`AbstractExecutorService`对`submit()`实现如下。

public Future<?> submit(Runnable task) {
 
 	if (task == null) throw new NullPointerException();
 	RunnableFuture<Void> ftask = newTaskFor(task, null);
 	execute(ftask);
 	return ftask;
 }
 
 public <T> Future<T> submit(Runnable task, T result) {
 
 	if (task == null) throw new NullPointerException();
 	RunnableFuture<T> ftask = newTaskFor(task, result);
 	execute(ftask);
 	return ftask;
 }
 
 public <T> Future<T> submit(Callable<T> task) {
 
 	if (task == null) throw new NullPointerException();
 	RunnableFuture<T> ftask = newTaskFor(task);
 	execute(ftask);
 	return ftask;
 }

在`execute()`方法中会根据当前线程数决定是新建线程来处理任务还是添加任务到任务阻塞队列中，而在`submit()`方法中是将任务封装成`RunnableFuture`然后再调用`execute()`方法。

#### 四. 关闭线程池 ####

可以通过调用线程池的`shutdown()`或者`shutdownNow()`方法来关闭线程池。

`shutdown()`方法会将线程池状态置为`SHUTDOWN`，此时线程池不会再接收新提交的任务，空闲的线程会被中断，当正在被执行的任务和任务阻塞队列中的任务执行完后线程池才会安全的关闭掉。

`shutdownNow()`方法会将线程池状态置为`STOP`，此时线程池不会再接收新提交的任务，所有线程会被中断，任务阻塞队列中的任务不再执行（这些任务会以列表形式返回），正在执行中的任务也会被尝试停止。

补充：上述中的空闲线程可以理解为正在从任务阻塞队列中获取任务的线程，即没有在执行任务的线程。

### 总结 ###

线程池的使用主要聚焦于线程池的参数的配置，而要合理的配置线程池的参数，则需要对线程池的原理有一定的了解。下篇文章将对线程池的原理进行深入分析。

[71fc607543f6407f84f1ebc09b554274.jpeg_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/f0f02c481bbe45e3a3282b94e3172e7a.jpeg
[a96ebb4b63b646579dbfaf11e48e4b40.jpeg_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/68eea140f3a049d6b39a463cec918de2.jpeg