[TOC]
# 1.ThreadPoolExecutor
> 1. 在《阿里巴巴java开发手册》中指出了线程资源必须通过线程池提供，不允许在应用中自行显示的创建线程，这样一方面是线程的创建更加规范，可以合理控制开辟线程的数量；
> 2. 线程池不允许使用Executors去创建，而要通过ThreadPoolExecutor方式，这一方面是由于jdk中Executor框架虽然提供了如newFixedThreadPool()、newSingleThreadExecutor()、newCachedThreadPool()等创建线程池的方法，但都有其局限性，不够灵活；另外由于前面几种方法内部也是通过ThreadPoolExecutor方式实现，使用ThreadPoolExecutor有助于大家明确线程池的运行规则，创建符合自己的业务场景需要的线程池，避免资源耗尽的风险。
## 1.1 构造函数详解
```
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
if (corePoolSize < 0 ||
maximumPoolSize <= 0 ||
maximumPoolSize < corePoolSize ||
keepAliveTime < 0)
throw new IllegalArgumentException();
if (workQueue == null || threadFactory == null || handler == null)
throw new NullPointerException();
this.acc = System.getSecurityManager() == null ?
null :
AccessController.getContext();
this.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}
```
**1. corePoolSize:**
指定了线程池中的线程数量，它的数量决定了添加的任务是开辟新的线程去执行，还是放到`workQueue`任务队列中去；
**2. maximumPoolSize:**
**指定了线程池中的最大线程数量**，这个参数会根据你使用的`workQueue`任务队列的类型，决定线程池会开辟的最大线程数量；
**3. keepAliveTime:**
当线程池中空闲线程数量超过corePoolSize时，多余的线程会在多长时间内被销毁；
**4. unit:** keepAliveTime的单位
**5. workQueue:**
任务队列，被添加到线程池中，但尚未被执行的任务；它一般分为直接提交队列、有界任务队列、无界任务队列、优先任务队列几种；
**6. threadFactory:** 线程工厂，用于创建线程，一般用默认即可；
**7. handler:** 拒绝策略；当任务太多来不及处理时，如何拒绝任务；
## 1.2 线程创建规则
1. 待执行任务数小于corePoolSize则创建线程，直到线程数等于corePoolSize
2. 待执行任务书大于corePoolSize，往队列里塞（如果队列无界，则线程数量永远等于corePoolSize）
3. 队列满了，创建新线程，直到线程数等于maximumPoolSize
4. 线程数量大于maximumPoolSize，执行拒绝策略
# 2. **使用**
上面我们已经介绍过了，**它一般分为直接提交队列、有界任务队列、无界任务队列、优先任务队列；**
## 2.1 **直接提交队列**：
设置为SynchronousQueue队列，**SynchronousQueue是一个特殊的BlockingQueue，它没有容量**，没执行一个插入操作就会阻塞，需要再执行一个删除操作才会被唤醒，反之每一个删除操作也都要等待对应的插入操作。
如下代码，最多创建三个线程
~~~
public class ThreadPool {
private static ExecutorService pool;
public static void main( String[] args )
{
//maximumPoolSize设置为2 ，拒绝策略为AbortPolic策略，直接抛出异常
pool = new ThreadPoolExecutor(0, 3, 1000, TimeUnit.MILLISECONDS, new SynchronousQueue<Runnable>(), Executors.defaultThreadFactory(),new ThreadPoolExecutor.AbortPolicy());
for(int i=0;i<5;i++) {
pool.execute(new ThreadTask());
}
}
}
class ThreadTask implements Runnable{
public ThreadTask() {
}
public void run() {
System.out.println(Thread.currentThread().getName());
}
}
~~~
输出：
```
[INFO] --- exec-maven-plugin:3.0.0:exec (default-cli) @ javatest ---
pool-1-thread-1
pool-1-thread-3
pool-1-thread-2
Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task com.tuna.test.thread.ThreadTask@33909752 rejected from java.util.concurrent.ThreadPoolExecutor@55f96302[Running, pool size = 3, active threads = 0, queued tasks = 0, completed tasks = 3]
at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2063)
at com.tuna.test.thread.ThreadPool.main(ThreadPool.java:12)
[ERROR] Command execution failed.
org.apache.commons.exec.ExecuteException: Process exited with an error: 1 (Exit value: 1)
at org.apache.commons.exec.DefaultExecutor.executeInternal (DefaultExecutor.java:404)
der.singlethreaded.SingleThreadedBuilder.build (SingleThreadedBuilder.java:56)
at org.codehaus.classworlds.Launcher.main (Launcher.java:47)
```
可以看到，当任务队列为SynchronousQueue，创建的线程数大于maximumPoolSize时，直接执行了拒绝策略抛出异常。
**使用SynchronousQueue队列，提交的任务不会被保存，总是会马上提交执行。** 如果用于执行任务的线程数量小于maximumPoolSize，则尝试创建新的进程，如果达到maximumPoolSize设置的最大值，则根据你设置的handler执行拒绝策略。因此这种方式你提交的任务不会被缓存起来，而是会被马上执行，在这种情况下，你需要对你程序的并发量有个准确的评估，才能设置合适的maximumPoolSize数量，否则很容易就会执行拒绝策略；
## 2.2 **有界的任务队列**：
有界的任务队列可以使用ArrayBlockingQueue实现，如下所示
~~~
pool = new ThreadPoolExecutor(1, 2, 1000, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(10),Executors.defaultThreadFactory(),new ThreadPoolExecutor.AbortPolicy());
~~~
1. 使用ArrayBlockingQueue有界任务队列，若有新的任务需要执行时，线程池会创建新的线程，直到创建的线程数量达到corePoolSize时，则会将新的任务加入到等待队列中。
2. 若等待队列已满，即超过ArrayBlockingQueue初始化的容量，则继续创建线程，直到线程数量达到maximumPoolSize设置的最大线程数量，若大于maximumPoolSize，则执行拒绝策略。
3. 在这种情况下，线程数量的上限与有界任务队列的状态有直接关系，如果有界队列初始容量较大或者没有达到超负荷的状态，线程数将一直维持在corePoolSize以下，反之当任务队列已满时，则会以maximumPoolSize为最大线程数上限。
## 2.3 **无界的任务队列**
有界任务队列可以使用LinkedBlockingQueue实现，如下所示
~~~
pool = new ThreadPoolExecutor(1, 2, 1000, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(),Executors.defaultThreadFactory(),new ThreadPoolExecutor.AbortPolicy());
~~~
使用无界任务队列，线程池的任务队列可以无限制的添加新的任务，而线程池创建的最大线程数量就是你corePoolSize设置的数量，也就是说在这种情况下maximumPoolSize这个参数是无效的，哪怕你的任务队列中缓存了很多未执行的任务，当线程池的线程数达到corePoolSize后，就不会再增加了；若后续有新的任务加入，则直接进入队列等待，当使用这种任务队列模式时，一定要注意你任务提交与处理之间的协调与控制，不然会出现队列中的任务由于无法及时处理导致一直增长，直到最后资源耗尽的问题。
## 2.4 **优先任务队列：**
优先任务队列通过PriorityBlockingQueue实现，下面我们通过一个例子演示下
~~~
public class ThreadPool {
private static ExecutorService pool;
public static void main( String[] args )
{
//优先任务队列
pool = new ThreadPoolExecutor(1, 2, 1000, TimeUnit.MILLISECONDS, new PriorityBlockingQueue<Runnable>(),Executors.defaultThreadFactory(),new ThreadPoolExecutor.AbortPolicy());
for(int i=0;i<20;i++) {
pool.execute(new ThreadTask(i));
}
}
}
public class ThreadTask implements Runnable,Comparable<ThreadTask>{
private int priority;
public int getPriority() {
return priority;
}
public void setPriority(int priority) {
this.priority = priority;
}
public ThreadTask() {
}
public ThreadTask(int priority) {
this.priority = priority;
}
//当前对象和其他对象做比较，当前优先级大就返回-1，优先级小就返回1,值越小优先级越高
public int compareTo(ThreadTask o) {
return this.priority>o.priority?-1:1;
}
public void run() {
try {
//让线程阻塞，使后续任务进入缓存队列
Thread.sleep(1000);
System.out.println("priority:"+this.priority+",ThreadName:"+Thread.currentThread().getName());
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
~~~
我们来看下执行的结果情况
~~~
priority:0,ThreadName:pool-1-thread-1
priority:9,ThreadName:pool-1-thread-1
priority:8,ThreadName:pool-1-thread-1
priority:7,ThreadName:pool-1-thread-1
priority:6,ThreadName:pool-1-thread-1
priority:5,ThreadName:pool-1-thread-1
priority:4,ThreadName:pool-1-thread-1
priority:3,ThreadName:pool-1-thread-1
priority:2,ThreadName:pool-1-thread-1
priority:1,ThreadName:pool-1-thread-1
~~~
到除了第一个任务直接创建线程执行外，其他的任务都被放入了优先任务队列，按优先级进行了重新排列执行，且线程池的线程数一直为corePoolSize，也就是只有一个。
通过运行的代码我们可以看出PriorityBlockingQueue它其实是一个特殊的无界队列，它其中无论添加了多少个任务，线程池创建的线程数也不会超过corePoolSize的数量，只不过其他队列一般是按照先进先出的规则处理任务，而PriorityBlockingQueue队列可以自定义规则根据任务的优先级顺序先后执行。
## **2.5 拒绝策略**
一般我们创建线程池时，为防止资源被耗尽，任务队列都会选择创建有界任务队列，但种模式下如果出现任务队列已满且线程池创建的线程数达到你设置的最大线程数时，这时就需要你指定ThreadPoolExecutor的RejectedExecutionHandler参数即合理的拒绝策略，来处理线程池"超载"的情况。ThreadPoolExecutor自带的拒绝策略如下：
**1、AbortPolicy策略：该策略会直接抛出异常，阻止系统正常工作；**
**2、CallerRunsPolicy策略：如果线程池的线程数量达到上限，该策略会把任务队列中的任务放在调用者线程当中运行；**
**3、DiscardOledestPolicy策略：该策略会丢弃任务队列中最老的一个任务，也就是当前任务队列中最先被添加进去的，马上要被执行的那个任务，并尝试再次提交；**
**4、DiscardPolicy策略：该策略会默默丢弃无法处理的任务，不予任何处理。当然使用此策略，业务场景中需允许任务的丢失；**
**以上内置的策略均实现了**RejectedExecutionHandler接口，**当然你也可以自己扩展RejectedExecutionHandler接口，定义自己的拒绝策略，我们看下示例代码：**
如果任务数量超过了 1、corePoolSize 2. 队列容量 3.最大线程数 ---> 执行拒绝策略：抛异常、丢弃、再次尝试和自定义策略
线程池最大任务数 maxPoolSize(包含corePoolSize) + queneSize 如下能容纳的最大任务数量 7，同时执行的任务数 2
~~~
public class ThreadPool {
private static ExecutorService pool;
public static void main( String[] args )
{
//自定义拒绝策略
pool = new ThreadPoolExecutor(1, 2, 1000, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(5),
Executors.defaultThreadFactory(), new RejectedExecutionHandler() {
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
System.out.println(r.toString()+"执行了拒绝策略");
}
});
for(int i=0;i<10;i++) {
pool.execute(new ThreadTask());
}
}
}
public class ThreadTask implements Runnable{
public void run() {
try {
//让线程阻塞，使后续任务进入缓存队列
Thread.sleep(1000);
System.out.println("ThreadName:"+Thread.currentThread().getName());
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
~~~
输出结果：
~~~
com.hhxx.test.ThreadTask@33909752执行了拒绝策略
com.hhxx.test.ThreadTask@55f96302执行了拒绝策略
com.hhxx.test.ThreadTask@3d4eac69执行了拒绝策略
ThreadName:pool-1-thread-2
ThreadName:pool-1-thread-1
ThreadName:pool-1-thread-1
ThreadName:pool-1-thread-2
ThreadName:pool-1-thread-1
ThreadName:pool-1-thread-2
ThreadName:pool-1-thread-1
~~~
可以看到由于任务加了休眠阻塞，执行需要花费一定时间，导致会有一定的任务被丢弃，从而执行自定义的拒绝策略；
## **2.6 ThreadFactory自定义线程创建**
线程池中线程就是通过ThreadPoolExecutor中的ThreadFactory，线程工厂创建的。那么通过自定义ThreadFactory，可以按需要对线程池中创建的线程进行一些特殊的设置，如命名、优先级等，下面代码我们通过ThreadFactory对线程池中创建的线程进行记录与命名
~~~
public class ThreadPool {
private static ExecutorService pool;
public static void main( String[] args )
{
//自定义线程工厂
pool = new ThreadPoolExecutor(2, 4, 1000, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(5),
new ThreadFactory() {
public Thread newThread(Runnable r) {
System.out.println("线程"+r.hashCode()+"创建");
//线程命名
Thread th = new Thread(r,"threadPool"+r.hashCode());
return th;
}
}, new ThreadPoolExecutor.CallerRunsPolicy());
for(int i=0;i<10;i++) {
pool.execute(new ThreadTask());
}
}
}
public class ThreadTask implements Runnable{
public void run() {
//输出执行线程的名称
System.out.println("ThreadName:"+Thread.currentThread().getName());
}
}
~~~
我们看下输出结果
~~~
线程118352462创建
线程1550089733创建
线程865113938创建
ThreadName:threadPool1550089733
ThreadName:threadPool118352462
线程1442407170创建
ThreadName:threadPool1550089733
ThreadName:threadPool1550089733
ThreadName:threadPool1550089733
ThreadName:threadPool865113938
ThreadName:threadPool865113938
ThreadName:threadPool118352462
ThreadName:threadPool1550089733
ThreadName:threadPool1442407170
~~~
可以看到线程池中，每个线程的创建我们都进行了记录输出与命名。
# **3. ThreadPoolExecutor扩展**
ThreadPoolExecutor扩展主要是围绕beforeExecute()、afterExecute()和terminated()三个接口实现的，
**1、beforeExecute：线程池中任务运行前执行**
**2、afterExecute：线程池中任务运行完毕后执行**
**3、terminated：线程池退出后执行**
通过这三个接口我们可以监控每个任务的开始和结束时间，或者其他一些功能。下面我们可以通过代码实现一下
~~~
public class ThreadPool {
private static ExecutorService pool;
public static void main( String[] args ) throws InterruptedException
{
//实现自定义接口
pool = new ThreadPoolExecutor(2, 4, 1000, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(5),
new ThreadFactory() {
public Thread newThread(Runnable r) {
System.out.println("线程"+r.hashCode()+"创建");
//线程命名
Thread th = new Thread(r,"threadPool"+r.hashCode());
return th;
}
}, new ThreadPoolExecutor.CallerRunsPolicy()) {
protected void beforeExecute(Thread t,Runnable r) {
System.out.println("准备执行："+ ((ThreadTask)r).getTaskName());
}
protected void afterExecute(Runnable r,Throwable t) {
System.out.println("执行完毕："+((ThreadTask)r).getTaskName());
}
protected void terminated() {
System.out.println("线程池退出");
}
};
for(int i=0;i<10;i++) {
pool.execute(new ThreadTask("Task"+i));
}
pool.shutdown();
}
}
public class ThreadTask implements Runnable{
private String taskName;
public String getTaskName() {
return taskName;
}
public void setTaskName(String taskName) {
this.taskName = taskName;
}
public ThreadTask(String name) {
this.setTaskName(name);
}
public void run() {
//输出执行线程的名称
System.out.println("TaskName"+this.getTaskName()+"---ThreadName:"+Thread.currentThread().getName());
}
}
~~~
输出结果
~~~
线程118352462创建
线程1550089733创建
准备执行：Task0
准备执行：Task1
TaskNameTask0---ThreadName:threadPool118352462
线程865113938创建
执行完毕：Task0
TaskNameTask1---ThreadName:threadPool1550089733
执行完毕：Task1
准备执行：Task3
TaskNameTask3---ThreadName:threadPool1550089733
执行完毕：Task3
准备执行：Task2
准备执行：Task4
TaskNameTask4---ThreadName:threadPool1550089733
执行完毕：Task4
准备执行：Task5
TaskNameTask5---ThreadName:threadPool1550089733
执行完毕：Task5
准备执行：Task6
TaskNameTask6---ThreadName:threadPool1550089733
执行完毕：Task6
准备执行：Task8
TaskNameTask8---ThreadName:threadPool1550089733
执行完毕：Task8
准备执行：Task9
TaskNameTask9---ThreadName:threadPool1550089733
准备执行：Task7
执行完毕：Task9
TaskNameTask2---ThreadName:threadPool118352462
TaskNameTask7---ThreadName:threadPool865113938
执行完毕：Task7
执行完毕：Task2
线程池退出
~~~
可以看到通过对beforeExecute()、afterExecute()和terminated()的实现，我们对线程池中线程的运行状态进行了监控，在其执行前后输出了相关打印信息。另外使用shutdown方法可以比较安全的关闭线程池， 当线程池调用该方法后，线程池中不再接受后续添加的任务。但是，此时线程池不会立刻退出，直到添加到线程池中的任务都已经处理完成，才会退出。
# **4. 线程池线程数量**
线程吃线程数量的设置没有一个明确的指标，根据实际情况，只要不是设置的偏大和偏小都问题不大，结合下面这个公式即可
~~~
/**
* Nthreads=CPU数量
* Ucpu=目标CPU的使用率，0<=Ucpu<=1
* W/C=任务等待时间与任务计算时间的比率
*/
Nthreads = Ncpu*Ucpu*(1+W/C)
~~~
- 计算机网络
- 基础_01
- tcp/ip
- http转https
- Let's Encrypt免费ssl证书（基于haproxy负载）
- what's the http?
- 网关
- 网络IO
- http
- 工具
- Git
- 初始本地仓库并上传
- git保存密码
- Gitflow
- maven
- 1.生命周期命令
- 聚合与继承
- 插件管理
- assembly
- 资源管理插件
- 依赖范围
- 分环境打包
- dependencyManagement
- 版本分类
- 找不到主类
- 无法加载主类
- 私服
- svn
- gradle
- 手动引入第三方jar包
- 打包exe文件
- Windows
- java
- 设计模式
- 七大原则
- 1.开闭原则
- 2. 里式替换原则
- 3. 依赖倒置原则
- 4. 单一职责原则
- 单例模式
- 工厂模式
- 简单工厂
- 工厂方法模式
- 抽象工厂模式
- 观察者模式
- 适配器模式
- 建造者模式
- 代理模式
- 适配器模式
- 命令模式
- json
- jackson
- poi
- excel
- easy-poi
- 规则
- 模板
- 合并单元格
- word
- 读取
- java基础
- 类路径与jar
- 访问控制权限
- 类加载
- 注解
- 异常处理
- String不可变
- 跨域
- transient关键字
- 二进制编码
- 泛型1
- 与或非
- final详解
- Java -jar
- 正则
- 读取jar
- map
- map计算
- hashcode计算原理
- 枚举
- 序列化
- URLClassLoader
- 环境变量和系统变量
- java高级
- java8
- 1.Lambda表达式和函数式接口
- 2.接口的默认方法和静态方法
- 3.方法引用
- 4.重复注解
- 5.类型推断
- 6.拓宽注解的应用场景
- java7-自动关闭资源机制
- 泛型
- stream
- 时区的正确理解
- StringJoiner字符串拼接
- 注解
- @RequestParam和@RequestBody的区别
- 多线程
- 概念
- 线程实现方法
- 守护线程
- 线程阻塞
- 笔试题
- 类加载
- FutureTask和Future
- 线程池
- 同步与异步
- 高效简洁的代码
- IO
- ThreadLocal
- IO
- NIO
- 图片操作
- KeyTool生成证书
- 压缩图片
- restful
- 分布式session
- app保持session
- ClassLoader.getResources 能搜索到的资源路径
- java开发规范
- jvm
- 高并发
- netty
- 多线程与多路复用
- 异步与事件驱动
- 五种IO模型
- copy on write
- code style
- 布隆过滤器
- 笔试
- 数据库
- mybatis
- mybatis与springboot整合配置
- pagehelper
- 分页数据重复问题
- Java与数据库之间映射
- 拦截器
- 拦截器应用
- jvm
- 堆内存测试
- 线程栈
- 直接内存
- 内存结构
- 内存模型
- 垃圾回收
- 调优
- 符号引用
- 运行参数
- 方法区
- 分带回收理论
- 快捷开发
- idea插件
- 注释模板
- git
- pull冲突
- push冲突
- Excel处理
- 图片处理
- 合并单元格
- easypoi
- 模板处理
- 响应式编程
- reactor
- reactor基础
- jingyan
- 规范
- 数据库