Java-concurrent之ExecutorService

为了能够更好地控制多线程，JDK提供了一套Executor框架，帮助开发人员有效地进行线程控制，其本质就是一个线程池。

1. 概述

JDK5之后把工作单元和执行机制区分开了，工作单元包括 Runnable 和 Callable<T> ，而执行机制则由 Executor 框架提供。
Executor 框架为线程的启动、执行和关闭提供了便利，底层使用线程池实现。
使用 Executor 框架管理线程的好处在于简化管理、提高效率，还能避免this逃逸问题——是指不完整的对象被线程调用。
Executor 框架使用了两级调度模型进行线程的调度。
1. 在上层，Java多线程程序通常把应用分解为多个任务，然后使用用户调度框架 Executor 将这些任务映射为固定数量的线程；
2. 在底层，操作系统内核将这些线程映射到硬件处理器上。

2. 相关类型

主要类以及相关的继承体系如下：

Executor
    ExecutorService
        AbstractExecutorService
            ThreadPoolExecutor
            ForkJoinPool 【work-stealing模式】
        ScheduledExecutorService
            ScheduledThreadPoolExecutor

这些成员都位于 java.util.concurrent package中，属于JDK并发包中的核心类。
其中 ThreadPoolExecutor 表示一个线程池，提供了大量配置参数来覆盖尽可能广的需求。（可以看到工具类 Executors 中构建的各类线程池时，其底层都是使用的 ThreadPoolExecutor 或其子类。）
官方提供的工具类 Executors 所创建的各类线程池基本已经能够满足绝大部分需求。

3. `ThreadPoolExecutor`

在平时的开发中，我们最常接触到的应该就是ThreadPoolExecutor了。

3.1 构造函数

作为

ExecutorService

接口的实现类，Executor框架的最核心的类就是它了，而工具类

Executors

中提供的各类简化线程池创建的方法，其底层实现大部分都是使用的

ThreadPoolExecutor

或其子类来完成的。因此了解

ThreadPoolExecutor

对我们熟练掌握Java多线程编程有着非常重要的意义。

ThreadPoolExecutor

提供了多个构造函数，这里我们仅列举最具有代表性的一个：

public ThreadPoolExecutor(int corePoolSize, // 线程池中的核心线程数量，可以理解为最小线程数
				  int maximumPoolSize, // 最大线程数量
				  long keepAliveTime, // 线程允许空闲的最长时间
				  TimeUnit unit, // keepAliveTime参数的单位
				  BlockingQueue<Runnable> workQueue, // 已提交但还未分配到执行线程的任务构成的队列
				  ThreadFactory threadFactory, // 线程工厂实例，用途不言自明
				  RejectedExecutionHandler handler) { ... } // 在因线程数量和任务队列都达到上限而导致新加入的任务无处可去时，所采取的拒绝策略； JDK默认提供了四种策略。

通过以上的注释，相信读者应该能够比较清楚地了解

ThreadPoolExecutor

中几个重要参数的含义。

3.2 阻塞队列 `BlockingQueue<T>`

在

ThreadPoolExecutor

构造函数中传入的

BlockingQueue<T>

实例，其主要作用是用来存储已被提交但未能马上分配到线程去执行的任务。

通过搜索接口的继承链，我们发现JDK提供了如下这四种队列实现类：

3.2.1

ArrayBlockingQueue

有界的任务队列。有界正是该队列最大的特点；该类的实例在构建时，必须传入一个代表该队列最大容量的参数。

如果使用该队列实例作为任务队列，只有在其装载的任务满载时，才有可能将线程数提升到corePoolSize之上；换而言之就是除非系统非常繁忙，否则将确保核心线程数维持在corePoolSize。

关于使用该队列作为任务队列的线程池执行逻辑，请参阅下面的贴图。

3.2.2

LinkedBlockingQueue

无界的任务队列。该队列的特点就是除非系统资源耗尽，否则无界任务队列将不会出现任务入队失败的情况。

按照下面贴图中的执行逻辑，当使用此队列作为任务队列时，线程池中线程的数量将一直保持在corePoolSize；如果任务创建和处理的速度差异很大，无界队列将会保持快速增长，直到耗尽系统资源。

3.2.3

PriorityBlockingQueue

带有执行优先级的任务队列。这是一个特殊的无界队列，前面两种队列都是按照FIFO算法来处理任务的，但该队列可以根据任务自身额优先级顺序先后执行，这就确保了紧急任务优先执行，进而保证了质量。

3.2.4

SynchronousQueue

直接提交的队列。这个队列非常特殊，其并没有容量，每一次插入操作必须等待一个相应的删除操作；反之，每一个删除操作都要等待对应的插入操作。

如果使用本实例来作为任务队列，提交的任务将不会被真实地保存，而总是被直接提交给线程去执行；如果没有空闲线程，将尝试创建新的线程，如果线程数量已经达到最大值，将执行拒绝策略。

因此选择使用本类作为任务队列时，通常需要设置很大的

maximumPoolSize

值，否则拒绝策略很可能频繁执行。

3.3 线程工厂 `ThreadFactory`

讲完了

BlockingQueue<T>

，我们接着来看看第二个参数，也就是用于创建线程的线程工厂。

ThreadPoolExecutor

默认使用

Executors.defaultThreadFactory()

来作为线程工厂。这其中一大问题是线程名称过于泛化，导致问题排查时举步维艰（pool-1-thread-1的线程名格式，我们基本无法从中得到任何有用的信息）。

通过提供自定义的

ThreadFactory

我们将可以得到很多好处——设置更佳人性化的线程名来追踪问题，跟踪线程状态，优化线程池以满足业务场景，获取更详尽的线程堆栈信息以排错。

3.4 拒绝策略 `RejectedExecutionHandler`

作为前文我们提到的

ThreadPoolExecutor

构造函数中的最后一个参数类型。它代表了在任务数量超过了系统实际承载能力时，该如何处理？即当线程数量达到上限，同时任务队列中无法塞入更多的新任务时，我们需要一套策略来处理这类情况。

JDK默认提供了四种拒绝策略：

AbortPolicy 直接抛出异常，阻止系统正常工作。
CallerRunsPolicy 只要线程池未关闭，将直接在调用者线程中运行当前被丢弃的任务。
DiscardOldestPolicy 丢弃最老的那个请求（也就是下一个将要被执行的任务），并尝试再次提交当前任务。
DiscardPolicy 默默丢弃无法处理的任务，不进行任何操作。如果允许任务丢失，这应该是最好的一种方案了！

以上这四种策略都是作为

ThreadPoolExecutor

内部类而存在的。

3.5 任务调度逻辑

ThreadPoolExecutor.execute() 方法体现出来的任务调度逻辑如下

Java-concurrent之ExecutorService

3.6 `ThreadPoolExecutor` 扩展

ThreadPoolExecutor

提供了

beforeExecute()

，

afterExecute()

，

terminated()

三个接口来允许外界对线程池进行控制。

以上这三个方法都是 protected 访问级别，并且都是空实现；摆明了是交给扩展子类去实现的。
其中前两个方法被回调于 ThreadPoolExecutor.Worker.runWorker() 方法中。 ThreadPoolExecutor 中的工作线程正是 ThreadPoolExecutor.Worker 实例， Worker.runWorker() 会同时被多个线程访问，因此 beforeExecute() ， afterExecute() 也将被多线程访问，这一点需要记住。

@Test
public void customExtend() throws Exception {
	ExecutorService es = new MyThreadPoolExecutor(5, 5, 0L, TimeUnit.MILLISECONDS, CollUtil
			.<Runnable> newBlockingQueue(1, true));

	for (int i = 0; i < 5; i++) {
		MyTask myTask = new MyTask("LQ-" + i);
		es.execute(myTask);
		ThreadUtil.safeSleep(10);
	}

	es.shutdown();
	es.awaitTermination(1000, TimeUnit.MILLISECONDS);
}

private static class MyTask implements Runnable {

	private String name;

	public MyTask(String name) {
		this.name = name;
	}

	@Override
	public void run() {
		Console.log("current task is {}, current thread is {} 正在执行...", name, Thread
				.currentThread().getName());
		ThreadUtil.safeSleep(100);
	}

}

private static class MyThreadPoolExecutor extends ThreadPoolExecutor {
	public MyThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime,
			TimeUnit unit, BlockingQueue<Runnable> workQueue) {
		super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
	}

	@Override
	protected void afterExecute(Runnable r, Throwable t) {
		Console.log("current task is {}, 执行结束", ((MyTask) r).name);
	}

	@Override
	protected void beforeExecute(Thread t, Runnable r) {
		Console.log("current task is {}, current thread is {} 准备执行", ((MyTask) r).name, t
				.getName());
	}

	@Override
	protected void terminated() {
		Console.log("线程池退出");
	}
}

// -------------------------------------- 输出如下
// 可以看到线程之间并不是连续的，比如 任务 LQ-0的执行
// 执行完成的先后顺序不等于提交的顺序，即使是同样的任务。
current task is LQ-0, current thread is pool-1-thread-1 准备执行
current task is LQ-1, current thread is pool-1-thread-2 准备执行
current task is LQ-1, current thread is pool-1-thread-2 正在执行...
current task is LQ-0, current thread is pool-1-thread-1 正在执行...
current task is LQ-2, current thread is pool-1-thread-3 准备执行
current task is LQ-2, current thread is pool-1-thread-3 正在执行...
current task is LQ-3, current thread is pool-1-thread-4 准备执行
current task is LQ-3, current thread is pool-1-thread-4 正在执行...
current task is LQ-4, current thread is pool-1-thread-5 准备执行
current task is LQ-4, current thread is pool-1-thread-5 正在执行...
current task is LQ-1, 执行结束
current task is LQ-2, 执行结束
current task is LQ-3, 执行结束
current task is LQ-0, 执行结束
current task is LQ-4, 执行结束
线程池退出

4. 工具类 `Executors` 中提供的几种线程池

说完了核心，我们再来看看JDK提供的工具类

Executors

中提供的几种线程池。

4.1 CachedThreadPool

工具类

Executors

提供的

newCachedThreadPool

，其底层正是使用了该

SynchronousQueue

队列来暂存任务。

使用工具类

Executors.newCachedThreadPool()

创建的线程池，其maximumPoolSize值为Integer.MAX_VALUE，这一点在选择线程池执行业务逻辑时千万要警醒。

按照上面对该队列的探讨，可以得出结论：该线程池适用于执行很多的短期异步任务的小程序，或者负载比较轻的服务器。

4.2 FixedThreadPool

工具类

Executors

提供的

newCachedThreadPool

，其底层正是使用了该

LinkedBlockingQueue

队列来暂存任务。

按照上文对该队列的探讨，可以得出结论：该线程池适用于为了满足管理资源的需求，而需要限制当前线程数量的应用场景，它适用于负载比较重的服务器。没有空闲线程时, 新的任务将被暂存在一个任务队列中。

4.3 SingleThreadExecutor

工具类

Executors

提供的

newFixedThreadPool

，其底层正是使用了该

LinkedBlockingQueue

队列来暂存任务，所以

按照上文对该队列的探讨，可以得出结论：该线程池适用于需要保证顺序地执行各个任务，并且在任意时间点不会有多个线程在活动的场景。

4.4 ScheduledThreadPoolExecutor

该线程池属于和

ThreadPoolExecutor

同一层次的

ExecutorService

实现类。该线程池支持延迟任务，以及定时及周期性的任务执行；多数情况下可用来替代Timer类。

ScheduledThreadPoolExecutor适用于需要在多个后台线程执行周期任务，同时为了满足资源管理需求需要限制后台线程数量的应用场景。

4.5 ForkJoinPool

该线程池也是属于和

ThreadPoolExecutor

同一层次的

ExecutorService

实现类。

该线程池的工作模式类似于

ThreadPoolExecutor

，但是使用了work-stealing模式，其会为线程池中的每个线程创建一个队列，从而用work-stealing(任务窃取)算法使得线程可以从其他线程队列里窃取任务来执行。这样就可以避免一部分线程无所事事，而另外一部分却是过度负载。

5. Links

《亿级流量网站架构核心技术》 P244
《Java高并发程序设计》 P95
Executor简介

Java-concurrent之ExecutorService

1. 概述

2. 相关类型

3. `ThreadPoolExecutor`

3.1 构造函数

3.2 阻塞队列 `BlockingQueue<T>`

3.3 线程工厂 `ThreadFactory`

3.4 拒绝策略 `RejectedExecutionHandler`

3.5 任务调度逻辑

3.6 `ThreadPoolExecutor` 扩展

4. 工具类 `Executors` 中提供的几种线程池

4.1 CachedThreadPool

4.2 FixedThreadPool

4.3 SingleThreadExecutor

4.4 ScheduledThreadPoolExecutor

4.5 ForkJoinPool

5. Links

继续阅读

关于Gradle配置的小结

Java小案例——随机数猜测随机数猜测

nginx location中斜线的位置的重要性

27 Best Free Eclipse Plug-ins for Java Developer to be ProductiveCode Quality PluginsText Editor PluginsDependency ManagementVersion Control Integration PluginsFramework Development Continuous Integration Related PluginsOther Utility Plugins

Java String.format方法的简单使用

neo4j之cypher使用文档

GitHub连夜封杀！这份阿里 10W 字内部 Java 字面试手册到底有多强？

spark/scala关于【资源文件】加载方法概述外部文件加载方案测试资源文件打包入jar包中小结

mybatis_入门程序Mybatis入门

AOP编程_Android优雅权限框架(1)概念基础，2021金三银四前言正文大纲正文

Effective Java 8:通用程序设计

OOM三种类型

工厂模式-三种类型

【递归】高效率求2的n次幂

win10本地scala和spark安装安装scala安装spark

scala (3) Function 和 Method

Java-concurrent之ExecutorService

1. 概述

2. 相关类型

3. ThreadPoolExecutor

3.1 构造函数

3.2 阻塞队列 BlockingQueue<T>

3.3 线程工厂 ThreadFactory

3.4 拒绝策略 RejectedExecutionHandler

3.5 任务调度逻辑

3.6 ThreadPoolExecutor 扩展

4. 工具类 Executors 中提供的几种线程池

4.1 CachedThreadPool

4.2 FixedThreadPool

4.3 SingleThreadExecutor

4.4 ScheduledThreadPoolExecutor

4.5 ForkJoinPool

5. Links

继续阅读

3. `ThreadPoolExecutor`

3.2 阻塞队列 `BlockingQueue<T>`

3.3 线程工厂 `ThreadFactory`

3.4 拒绝策略 `RejectedExecutionHandler`

3.6 `ThreadPoolExecutor` 扩展

4. 工具类 `Executors` 中提供的几种线程池