一个有趣的问题：如何设计一个线程池

理解Java并发工具包线程池的设计
深度解读 java 线程池设计思想及源码实现

分布式锁unlock 问题产生原因分析：

Step 1 ：线程A先上同一个锁（Key）（20秒），然后执行耗时业务，业务太长时间的执行（30秒，琐失效）；
与此同时，
Step 2 ：过了20秒，琐失效，线程B获取到该锁进行业务，A执行任务之后去释放该锁（锁被B持有了）
报 Exception in thread “thread-2” java.lang.IllegalMonitorStateException: attempt to unlock lock, not locked by current thread by node id: 9f178836-f7e1-44fe-a89d-2db52f399c0d thread-id: 21
at org.redisson.RedissonLock.unlock(RedissonLock.java:353) ，

为什么uat会出现，生产没出现呢？

   原因可能是uat 经常发版，生产执行很快，20秒的锁时间应该够了

第0个问题：怎么去运用线程池？工作中如何使用？

工作中，我们有时候需要实现一些耗时的任务，例如会有,将Word转换成PDF存储. 的需求

1 没使用线程池的实现代码–每次都开启新的线程,会导致资源耗尽，系统宕机

import org.junit.Test;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;public class ThreadVs {/*** 老的处理方式*/@Testpublic void oldHandle() throws InterruptedException {/*** 使用循环来模拟许多用户请求的场景*/for (int request = 1; request <= 100; request++) {new Thread(() -> {System.out.println("文档处理开始！");try {// 将Word转换为PDF格式：处理时长很长的耗时过程Thread.sleep(1000L * 30);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("文档处理结束！");}).start();}Thread.sleep(1000L * 1000);}
}

2 使用线程池的实现代码

 /*** 新的处理方式*/@Testpublic void newHandle() throws InterruptedException {/*** 开启了一个线程池：线程个数是10个*/ExecutorService threadPool =Executors.newFixedThreadPool(10);/*** 使用循环来模拟许多用户请求的场景*/for (int request = 1; request <= 100; request++) {threadPool.execute(() -> {System.out.println("文档处理开始！");try {// 将Word转换为PDF格式：处理时长很长的耗时过程Thread.sleep(1000L * 30);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("文档处理结束！");});}Thread.sleep(1000L * 1000);}

首先，我们做一件事情之前，总会有三个关键性的问题？

定义问题：是什么
问题的意义：为什么
解决问题：怎么做

第一个问题：线程池是什么？

线程池顾名思义就是事先创建若干个可执行的线程放入一个池中（容器），需要的时候从池中获取线程不用自行创建，使用完毕不需要销毁线程而是放回池中，从而减少创建和销毁线程对象的开销。任何池化技术都是减低资源消耗，例如我们常用的数据库连接池

第2个问题：为什么要用线程池？

降低资源消耗
提高响应速度
提高线程的可管理性

如果我们不使用已有的线程池, 而是我们自己去设计的话, 怎么去设计一个线程池？简单线程池设计？

设计过程中我们需要思考的问题

初始创建多少线程？
没有可用线程了怎么办？
缓冲数组需要设计多长？
缓冲数组满了怎么办？

第一次需求分析：简陋版本
下图是最简陋的线程池版本：具有的功能有

客户端获取线程
客户端归还线程
开启线程池，初始化线程池，关闭线程池

该设计方案,如下图所示,我们需要考虑下面几个问题：
1.在获取线程的时候，线程池没有线程可以获取的情况怎么处理?
2.初始化线程池时候，初始化多少个线程才算合适?
3.对于客户端使用不够方便，使用之后还要归还线程?不好使用

第二次需求分析：改进版本

改进版依然需要解决的三个问题

任务队列多长才好
队列满了之后怎么办？应该采取什么策略
线程池初始化，初始化多少线程才合适？
设计之前参考巨人的设计

corePoolSize 核心线程数量
maximumPoolSize 最大线程数量
keepAliveTime 线程空闲后的存活时间（没有任务后）
unit 时间单位
workQueue 用于存放任务的阻塞队列
threadFactory 线程工厂类
handler 当队列和最大线程池都满了之后的饱和策略

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.corePoolSize = corePoolSize;this.maximumPoolSize = maximumPoolSize;this.workQueue = workQueue;this.keepAliveTime = unit.toNanos(keepAliveTime);this.threadFactory = threadFactory;this.handler = handler;

java 源码里面设计的线程池的处理流程

注意1个问题：

阻塞队列未满，是不会创建新的线程的

线程池可选择的阻塞队列

插入移除操作：插入操作和移除操作

无界队列：无限长的队列阻塞队列，可以一直往里面追加元素 LinkedBlockingQueue
有界队列：有界限的阻塞队列，ArrayBlockingQueue
同步移交队列：不存储元素的阻塞队列，每个插入的操作必须等待另外一个线程取出元素，SynchronousQueue ，消费者生产者缓冲作用，RocketMQ

下面是三种阻塞队列的Java代码实现

import org.junit.Test;import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.SynchronousQueue;public class QueueTest {@Testpublic void arrayBlockingQueue() throws InterruptedException {/*** 基于数组的有界阻塞队列，队列容量为10*/ArrayBlockingQueue queue =new ArrayBlockingQueue<Integer>(10);// 循环向队列添加元素for (int i = 0; i < 20; i++) {queue.put(i);System.out.println("向队列中添加值：" + i);}}@Testpublic void linkedBlockingQueue() throws InterruptedException {/*** 基于链表的有界/无界阻塞队列，队列容量为10*/LinkedBlockingQueue queue =new LinkedBlockingQueue<Integer>();// 循环向队列添加元素for (int i = 0; i < 20; i++) {queue.put(i);System.out.println("向队列中添加值：" + i);}}@Testpublic void test() throws InterruptedException {/*** 同步移交阻塞队列*/SynchronousQueue queue = new SynchronousQueue<Integer>();// 插入值new Thread(() -> {try {queue.put(1);System.out.println("插入成功");} catch (InterruptedException e) {e.printStackTrace();}}).start();// 删除值/*new Thread(() -> {try {queue.take();System.out.println("删除成功");} catch (InterruptedException e) {e.printStackTrace();}}).start();*/Thread.sleep(1000L * 60);}}

线程池可选择的饱和策略
当阻塞队列满和最大线程数满了的时候，饱和策略就会发挥作用

AbortPolicy 终止策略（默认）: 通过抛出异常
DiscardPolicy ：丢弃策略：什么都不做
DiscardOldestPolicy : 丢弃旧任务策略：丢弃最久的任务，执行当前任务
CallerRunsPolicy ：调用者自运行策略：调用方自己执行自己的任务

线程池的执行示意图

第一步：主线程调用execute()方法来执行一个线程任务
第二步：如果核心线程池没有满，会立即创建新的线程来执行任务，如果核心线程池已经满了，则会调用方法2
第三步：当阻塞队列也和核心线程都满了之后，会执行方法3，从最大线程池数量里面获取线程，前提是不超过最大线程数
第四步：如果方法3也没法走通，接着执行方法4，执行饱和策略
第5步：如果饱和策略是 CallerRunsPolicy ，交给主线程自己去运行任务的run方法

常用线程池

newCachedThreadPool 线程数量无限大的，同步移交队列的线程池

 /*** Creates a thread pool that creates new threads as needed, but* will reuse previously constructed threads when they are* available.  These pools will typically improve the performance* of programs that execute many short-lived asynchronous tasks.* Calls to {@code execute} will reuse previously constructed* threads if available. If no existing thread is available, a new* thread will be created and added to the pool. Threads that have* not been used for sixty seconds are terminated and removed from* the cache. Thus, a pool that remains idle for long enough will* not consume any resources. Note that pools with similar* properties but different details (for example, timeout parameters)* may be created using {@link ThreadPoolExecutor} constructors.** @return the newly created thread pool*/// 线程数量无限大的线程池，需要小心public static ExecutorService newCachedThreadPool() {return new ThreadPoolExecutor(0, Integer.MAX_VALUE,60L, TimeUnit.SECONDS,new SynchronousQueue<Runnable>());}

 *创建一个线程池，该线程池根据需要创建新线程将重用先前构造的可用的线程。*这些池通常可以提高性能执行许多短暂的异步任务的程序。*调用{@code execute}将重用以前构造的线程（如果有）。* 如果没有现有线程可用，则新线程将被创建并添加到池中。*具有的线程*六十秒未使用将终止并从缓存中删除*因此，闲置足够长时间的池将不消耗任何资源。请注意，类似的池属性，*但细节不同（例如，超时参数）可以使用{@link ThreadPoolExecutor}构造函数创建。* @返回新创建的线程池

newFixedThreadPool 线程数量固定，无界阻塞队列的线程池

/**
* 线程数量固定的线程池
* nThreads 核心线程数和最大核心线程数
* LinkedBlockingQueue 无界阻塞队列，注意这个是无界的无限长的任务队列
*/public static ExecutorService newFixedThreadPool(int nThreads) {return new ThreadPoolExecutor(nThreads, nThreads,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>());}

newSingleThreadExecutor 线程数量只有1的无界阻塞队列线程池

/**
* 单一线程的线程池
* LinkedBlockingQueue 无界阻塞队列，注意这个是无界的无限长的任务队列
*/public static ExecutorService newSingleThreadExecutor() {return new FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>()));}

如何向线程池提交任务

向线程池提交任务的两种方式：

第一种：利用submit方法提交任务，接收任务的返回结果

@Testpublic void submitTest()throws ExecutionException, InterruptedException {// 创建线程池ExecutorService threadPool =Executors.newCachedThreadPool();/*** 利用submit方法提交任务，接收任务的返回结果*/Future<Integer> future = threadPool.submit(() -> {Thread.sleep(1000L * 10);return 2 * 5;});/*** 阻塞方法，直到任务有返回值后，才向下执行*/Integer num = future.get();System.out.println("执行结果：" + num);}

第二种：用execute方法提交任务，没有返回结果

@Testpublic void executeTest() throws InterruptedException {// 创建线程池ExecutorService threadPool =Executors.newCachedThreadPool();/*** 利用execute方法提交任务，没有返回结果*/threadPool.execute(() -> {try {Thread.sleep(1000L * 10);} catch (InterruptedException e) {e.printStackTrace();}Integer num = 2 * 5;System.out.println("执行结果：" + num);});Thread.sleep(1000L * 1000);}

线程池的状态