推荐阅读：Java线程池实现原理及其在美团业务中的实践

什么是线程池

• 线程池就是提前创建若干个线程，如果有任务需要处理，线程池里的线程就会处理任务，处理完之后线程并不会被销毁,而是继续放在线程池中，等待下一个任务。和连接池有点类似

使用线程池的好处

1. 降低资源消耗。通过重复利用已创建的线程降低线程的创建和销毁造成的消耗
2. 提高相应速度。当任务到达时，任务可疑不需要等到线程创建就能立即执行
3. 提高线程的可管理性。线程是稀缺资源，如果无限制的创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可疑进行同意分配、调优和监控

线程池的实现原理

当向线程池提交了一个任务后，线程池是如何处理这个任务的呢？
我们来看看这个流程图

从流程图中我们可以很清楚的看到

1. 首先有任务来会判断核心线程池里的线程是否已满（大于或等于corePoolSize）,如果不是，则创建一个新的线程来执行任务
2. 是，进入下一个流程判断判断工作队列是否已经满了。如果工作队列没满，则将任务放到工作队列中
3. 工作队列满了，判断线程池的线程是否都处理工作状态，是，则创建新的线程继续。如果创建的线程超过maximumPoolSize（线程池已满）。则交给饱和策略来处理这个任务

个人解释说明：就是首先有任务来就会在核心线程池创建线程，再来任务不管核心线程池中有没有线程空闲，都会继续创建线程，直到核心线程池满了，就会放到任务队列中，如果任务队列也满了，就会创建非核心线程，注意核心线程池的线程是不会死亡的，非核心到过期时间会被销毁

流程图分析

我们来看看 ThreadPoolExecutor执行execute

1. 如果当前运行的线程少于corePoolSize，则创建新线程来执行任务（注意，执行这一步骤
   需要获取全局锁）
2. 如果运行的线程等于或大于corePoolSize,则将任务加入到BlockingQueue
3. 如果BlockingQueue 队列已满，则创建新的线程来处理任务（执行这一步需要获取全局锁）
4. 如果创建新线程将使当前运行的线程超出 maximumPoolSize ，任务将被拒绝，并调用RejectedExecutionHandler.rejectedExecution()方法

ThreadPoolExecutor采取上述步骤的总体设计思路，是为了在执行execute()方法时，尽可能
地避免获取全局锁（那将会是一个严重的可伸缩瓶颈）。在ThreadPoolExecutor完成预热之后
（当前运行的线程数大于等于corePoolSize），几乎所有的execute()方法调用都是执行步骤2，而
步骤2不需要获取全局锁

源码分析

我们来看看 ThreadPoolExecutor 中 execute 方法的实现

public void execute(Runnable command) {if (command == null)throw new NullPointerException();int c = ctl.get();//1.当前池中线程比核心数少，新建一个线程执行任务if (workerCountOf(c) < corePoolSize) {   if (addWorker(command, true))return;c = ctl.get();}//2.核心池已满，但任务队列未满，添加到队列中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);}//3.核心池已满，队列已满，试着创建一个新线程else if (!addWorker(command, false))reject(command);    //如果创建新线程失败了，说明线程池被关闭或者线程池完全满了，拒绝任务
}

线程池中的线程执行任务分两种情况，如下

1. 在execute()方法中创建一个线程时，会让这个线程执行当前任务。
2. 这个线程执行完上图中1的任务后，会反复从BlockingQueue获取任务来执行

线程池的使用

现在常用的线程池分别是这四种：

1. newCachedThreadPool：可缓存线程池，先查看池中有没有以前建立的线程，如果有，就直接使用。如果没有，就建一个新的线程加入池中。线程池为无限大，当执行第二个任务时第一个任务已经完成，会复用执行第一个任务的线程，而不用每次新建线程
2. newFixedThreadPool：创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待
3. newScheduledThreadPool：创建一个定长线程池，支持定时及周期性任务执行
4. newSingleThreadExecutor：创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行

这四种线程池都是由Executor接口创建的，而Executor接口的最顶层实现都是ThreadPoolExecutor类，创建这些不同的线程池也就是调用ThreadPoolExecutor类的构造方法传入的不同而已，所以我们先来深入了解一下ThreadPoolExecutor的构造方法

 public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue) {this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,Executors.defaultThreadFactory(), defaultHandler);}public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory) {this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,threadFactory, defaultHandler);}
public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,RejectedExecutionHandler handler) {this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,Executors.defaultThreadFactory(), 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;}

• corePoolSize：核心线程池的大小。
• maximunPoolSize：线程池最大数量。线程池允许创建的最大线程数。如果队列满了，并
  且已创建的线程数小于最大线程数，则线程池会再创建新的线程执行任务。值得注意的是，如
  果使用了无界的任务队列这个参数就没什么效果
• keepAliveTime：非核心线程能够空闲的最长时间，超过时间，线程终止。果任务很多，并且每个任务执行的时间比较短，可以调大时间，提高线程的利用率
• unit：时间单位和keepAliveTime一起使用，可选的单位有天（DAYS）、小时（HOURS）、分钟
  （MINUTES）、毫秒（MILLISECONDS）、微秒（MICROSECONDS，千分之一毫秒）和纳秒（NANOSECONDS，千分之一微秒）
• workQueue：缓存队列，用来存放等待被执行的任务
• threadFactory：用于设置创建线程的工厂
• handler：饱和策略，线程数量大于最大线程数就会采用饱和策略，五种策略为：

1. AbortPolicy：直接抛出异常
2. CallerRunsPolicy：只用调用者所在线程来运行任务
3. DiscardOldestPolicy：丢弃队列里最近的一个任务，并执行当前任务
4. DiscardPolicy：不处理，丢弃掉
5. 根据应用场景需要来实现RejectedExecutionHandler接口自定义策略

向线程池中提交任务

1. execute()：用于提交不需要返回值的任务，所以无法判断任务是否被线程池执行成功,一般线程都是使用这个方法提交任务

void execute(Runnable command);

2. submit()：用于提交需要返回值的任务。线程池会返回一个future类型的对象，通过这个
   future对象可以判断任务是否执行成功，并且可以通过future的get()方法来获取返回值，get()方
   法会阻塞当前线程直到任务完成，而使用get（long timeout，TimeUnit unit）方法则会阻塞当前线
   程一段时间后立即返回，这时候有可能任务没有执行完
   

newCachedThreadPool

通过构造方法我们可以看到newCachedThreadPool线程池是无界的，60秒回收

public class Test {public static void main(String[] args) {ExecutorService newCachedThreadPool = Executors.newCachedThreadPool();for (int i = 0; i < 10; i++) {try {Thread.sleep(100);} catch (Exception e) {// TODO: handle exception}newCachedThreadPool.execute(()->{System.out.println(Thread.currentThread().getName() + "正在执行");});}//不熟悉lambda使用这种方式/* for (int i = 0; i < 10; i++) {newCachedThreadPool.execute(new Runnable() {@Overridepublic void run() {try {Thread.sleep(100);} catch (Exception e) {// TODO: handle exception}System.out.println(Thread.currentThread().getName() + "正在执行");}});}*/}
}

可以看到始终是线程1在被复用

newFixedThreadPool

构造函数

可以看到最大线程数与核心线程数保持一致这样就保证了传入的线程数一直被复用

public class Test {public static void main(String[] args) {ExecutorService newFixedThreadPool = Executors.newFixedThreadPool(2);}
}

可以看到始终只有两个线程池在执行

newScheduledThreadPool

构造方法

默认也是无界的

public class Test {public static void main(String[] args) {ScheduledExecutorService newScheduledThreadPool = Executors.newScheduledThreadPool(5);for (int i = 0; i < 10; i++) {newScheduledThreadPool.schedule(()-> System.out.println(Thread.currentThread().getName() + "延迟三秒执行"),3,TimeUnit.SECONDS);}//不熟悉lambda使用这种方式/* for (int i = 0; i < 10; i++) {newScheduledThreadPool.schedule(new Runnable() {public void run() {System.out.println(Thread.currentThread().getName() + "延迟三秒执行");}}, 3, TimeUnit.SECONDS);}*/}
}

这段任务在三秒后才会开始执行

如果要循环执行，只需要把这样既可以

newScheduledThreadPool .scheduleAtFixedRate(new Runnable() {
13             public void run() {System.out.println(Thread.currentThread().getName() + "延迟三秒执行后循环执行");
15             }
16         }, 3, 3, TimeUnit.SECONDS);

newSingleThreadExecutor

构造方法

可以看到核心线程数和最大线程数都为1,始终只有一个线程数

public class Test {public static void main(String[] args) {ExecutorService newSingleThreadExecutor = Executors.newSingleThreadExecutor();for (int i = 0; i < 10; i++) {final int index = i;newSingleThreadExecutor.execute(()->{System.out.println(Thread.currentThread().getName() +"打印的值为" + index);try {Thread.sleep(200);} catch (Exception e) {// TODO: handle exception}});//不熟悉lambda使用这种方式/*newSingleThreadExecutor.execute(new Runnable() {@Overridepublic void run() {System.out.println(Thread.currentThread().getName() +"打印的值为" + index);try {Thread.sleep(200);} catch (Exception e) {// TODO: handle exception}}});*/}}
}

可以看到始终只有一个线程，按顺序打印结果

自定义线程池

public class CustomThreadPool {public static void main(String[] args) {ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(1, 2, 60L, TimeUnit.SECONDS, new ArrayBlockingQueue<>(3));try {for (int i = 0; i < 6; i++) {TaskThread taskThread = new TaskThread("任务" + i);threadPoolExecutor.execute(taskThread);}} catch (Exception e) {threadPoolExecutor.shutdown();e.printStackTrace();}}}class TaskThread implements Runnable {private String taskName;public TaskThread(String taskName) {this.taskName = taskName;}@Overridepublic void run() {System.out.println(Thread.currentThread().getName()+taskName);}
}

这里可以看到队列满了的情况下并且超出最大线程数，队列就执行拒绝策略

Springboot 线程池的封装使用

@Component(ThreadPoolManager.PACKAGE_BEAN_NAME)
public class ThreadPoolManager implements DisposableBean {private static final Logger logger = LoggerFactory.getLogger(ThreadPoolManager.class);public static final String PACKAGE_BEAN_NAME = "ThreadPoolManager";/*** 定义线程池*/private ThreadPoolExecutor threadPool = new ThreadPoolExecutor(10, 500, 300, TimeUnit.SECONDS,new ArrayBlockingQueue<>(25),new ThreadFactoryBuilder().setDaemon(false).build(),new ThreadPoolExecutor.CallerRunsPolicy());/*** @param name* @param runnable*/public void execute(final String name, final Runnable runnable) {execute(name, runnable, null);}/*** @param name* @param runnable* @param exceptionHandler 异常处理*/public void execute(final String name, final Runnable runnable, final Thread.UncaughtExceptionHandler exceptionHandler) {threadPool.execute(() -> {final Thread currentThread = Thread.currentThread();setName(currentThread, name);if (Objects.isNull(exceptionHandler)) {//默认只打印异常日志currentThread.setUncaughtExceptionHandler((thread, exception) -> {logger.error("线程执行异常-{}", currentThread.getName(), exception);});} else {currentThread.setUncaughtExceptionHandler(exceptionHandler);}runnable.run();});}/*** 销毁线程池* @throws Exception*/@Overridepublic void destroy() throws Exception {MoreExecutors.shutdownAndAwaitTermination(threadPool, 20, TimeUnit.SECONDS);}/*** @param callable* @return submit方式提交任务，则异常不能被异常处理器捕获，如果一个由submit提交的任务由于抛出了异常而结束，那么这个异常将被Future.get封装在ExecutionException中重新抛出*/public <T> Future<T> submit(String name, final Callable<T> callable) {return threadPool.submit(() -> {setName(Thread.currentThread(), name);return callable.call();});}private void setName(Thread thread, String name) {thread.setName(name);}/*** 获取线程池配置* @return*/public JSONObject getConfig() {JSONObject data = new JSONObject();data.put("activeCount", threadPool.getActiveCount());data.put("completedTaskCount", threadPool.getCompletedTaskCount());data.put("queue.size()", threadPool.getQueue().size());data.put("taskCount", threadPool.getTaskCount());data.put("corePoolSize", threadPool.getCorePoolSize());data.put("largestPoolSize", threadPool.getLargestPoolSize());data.put("maximumPoolSize", threadPool.getMaximumPoolSize());return data;}
}