ThreadPoolExecutor 简介
ThreadPoolExecutor 是 java.util.concurrent 包下的一个类，在jdk1.5版本引入，帮助开发人员管理线程并方便地执行并行任务。
通俗来说，ThreadPoolExecutor 的作用是生产和管理线程池的，可以通过调用其 execute 方法和 submit 方法执行多线程任务。
ThreadPoolExecutor 使用
创建执行器
ExecutorService 对象和 ThreadPoolExecutor 的关系如下图：
ExecutorServiceConfig：
package com.ramble.threadpool.config;import java.util.concurrent.*; public class ExecutorServiceConfig { /** * 定义一个并发任务执行器服务 */ private static ExecutorService executorService; /** * 在类加载的时候初始化并发任务执行器 */ static { init(); } /** * 防止类属性被篡改 */ private ExecutorServiceConfig() { } /** * 初始化并发任务执行器。核心线程数量：设置为2，初始创建的线程池大小；最大线程数量：设置为3；空闲线程存活时间：设置为3秒，当非核心线程执行完任务之后，若没有新的任务分派，存活多久后自动销毁；任务队列：设置为2，当线程池创建的线程数量达到最大线程数量后，新进来的任务会排队等候； * 拒绝策略：设置为直接抛异常 * <p> * 以上配置需要根据：实际的业务场景、项目实际情况、实际硬件情况等各种因素综合考量 */ private static void init() { executorService = new ThreadPoolExecutor(2, 3, 3, TimeUnit.SECONDS, new LinkedBlockingDeque<Runnable>(2), new ThreadPoolExecutor.AbortPolicy()); } /** * 获取默认并发任务执行器 * * @return */ public static ExecutorService getDefaultExecutor() { return executorService; } /** * 获取固定大小并发任务执行器 * * @return */ public static ExecutorService getFixedExecutor() { return Executors.newFixedThreadPool(10); } /** * 获取其他并发任务执行器 * @return */ public static ExecutorService getOtherExecutor() { //todo return null; } } -
这个类的核心目的是构造一个 ExecutorService ， 供业务代码调用。当业务代码需要创建线程执行任务的时候，不用创建、管理和维护“执行者（线程）”，只需要告诉 ExecutorService 需要做什么 “事情（任务）”。
-
ExecutorService 对象必须是单例的，因为此对象本身是管理线程池的，如果自己都不是线程安全的，那使用起来将有可能发生灾难。
-
一个java进程允许创建多个ExecutorService 。根据业务实际情况，如果业务逻辑比较单一，大概率创建一种就满足使用。若业务繁杂，可根据业务特性抽象出多种类型以满足不同需求。譬如代码中就创建了两个 getDefaultExecutor 和 getFixedExecutor
-
关于线程池配置，如核心线程数量、最大线程数量、拒绝策略等等没有绝对正确的值做参考，需要根据实际情况设定
创建任务
- 通过实现 Runnable 接口并重写 run 方法创建无返回值的多线程任务
- 通过实现 Callable 接口并重写 call 方法创建有返回值的多线程任务
Task1:
package com.ramble.threadpool.task; import lombok.extern.slf4j.Slf4j; @Slf4j public class Task1 implements Runnable { /** * 新线程执行一个任务，没有参数，不需要返回值，此任务和其他任务没有先后顺序 */ @Override public void run() { try { Thread.sleep(3000L); log.info("TaskOne,thread is ={},thread name is={}", Thread.currentThread().getId(), Thread.currentThread().getName()); throw new IllegalAccessException("主动抛一个异常"); } catch (IllegalAccessException e) { throw new RuntimeException(e); } catch (InterruptedException e) { throw new RuntimeException(e); } } } Task2：
package com.ramble.threadpool.task; import com.alibaba.fastjson.JSON; import com.ramble.threadpool.dto.TaskTwoParam; import lombok.extern.slf4j.Slf4j; @Slf4j public class Task2 implements Runnable { private TaskTwoParam param; public Task2(TaskTwoParam param) { this.param = param; } /** * 新线程执行一个任务，参数通过构造函数传递，不需要返回值，此任务和其他任务没有先后顺序 */ @Override public void run() { try { Thread.sleep(3000L); } catch (InterruptedException e) { throw new RuntimeException(e); } log.info("TaskTwo,thread id is ={},thread name is={}, param is ={}", Thread.currentThread().getId(), Thread.currentThread().getName(), JSON.toJSONString(param)); } } Task3：
package com.ramble.threadpool.task; import com.alibaba.fastjson.JSON; import com.ramble.threadpool.dto.TaskDto; import com.ramble.threadpool.dto.TaskTwoParam; import lombok.extern.slf4j.Slf4j; import java.util.concurrent.Callable; @Slf4j public class Task3 implements Callable { private TaskTwoParam param; public Task3(TaskTwoParam param) { this.param = param; } /** * 新线程执行一个任务，通过构造函数传递参数，有返回值，此任务和其他任务没有先后顺序 * * @return * @throws Exception */ @Override public Object call() throws Exception { log.info("Task3,thread id is ={},thread name is={}, param is ={}", Thread.currentThread().getId(), Thread.currentThread().getName(), JSON.toJSONString(param)); return new TaskDto().setName("task3-callable"); } } Task4：
package com.ramble.threadpool.task; import lombok.extern.slf4j.Slf4j; @Slf4j public class Task4 implements Runnable { /** * 新线程执行一个任务，没有参数，不需要返回值，此任务和其他任务没有先后顺序 */ @Override public void run() { try { Thread.sleep(3000L); log.info("Task4,thread is ={},thread name is={}", Thread.currentThread().getId(), Thread.currentThread().getName()); } catch (InterruptedException e) { throw new RuntimeException(e); } } } 执行任务
在 controller 中模拟发起多线程任务。
package com.ramble.threadpool.controller; import com.alibaba.fastjson.JSON; import com.ramble.threadpool.config.ExecutorServiceConfig; import com.ramble.threadpool.dto.TaskDto; import com.ramble.threadpool.dto.TaskTwoParam; import com.ramble.threadpool.task.Task1; import com.ramble.threadpool.task.Task2; import com.ramble.threadpool.task.Task3; import com.ramble.threadpool.task.Task4; import lombok.extern.slf4j.Slf4j; import org.springframework.boot.autoconfigure.task.TaskExecutionProperties; import org.springframework.scheduling.annotation.Async; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; import javax.annotation.Resource; import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; @GetMapping("/task") public String testTask() { log.info("testTask,thread is ={},thread name is={}", Thread.currentThread().getId(), Thread.currentThread().getName()); // for (int i = 0; i < 4; i++) { // ExecutorServiceConfig.getExecutor().execute(new Task1()); // } // // for (int i = 0; i < 5; i++) { // ExecutorServiceConfig.getExecutor().execute(new Task2(new TaskTwoParam().setId(100).setName("cnaylor"))); // } ExecutorServiceConfig.getDefaultExecutor().execute(new Task1()); ExecutorServiceConfig.getDefaultExecutor().execute(new Task2(new TaskTwoParam().setId(100).setName("cnaylor"))); ExecutorServiceConfig.getFixedExecutor().execute(new Task4()); Future<TaskDto> taskResult = ExecutorServiceConfig.getDefaultExecutor().submit(new Task3(new TaskTwoParam().setId(100).setName("cnaylor"))); TaskDto taskDto; try { taskDto = taskResult.get(); } catch (InterruptedException e) { throw new RuntimeException(e); } catch (ExecutionException e) { throw new RuntimeException(e); } return "testTask"; } ThreadPoolExecutor 拒绝策略
所谓拒绝策略，一般是指当前线程池处于满负荷状态，所有的线程都有正在处理的任务，阻塞队列也排满了的情况下，对新进来的任务做出何种响应。
默认的拒绝策略有四种，同时也可以自定义拒绝策略。
- AbortPolicy
- CallerRunsPolicy
- DiscardOldestPolicy
- DiscardPolicy
AbortPolicy
抛一个异常（RejectedExecutionException），此任务将无法执行。
CallerRunsPolicy
线程池满载了， 此任务将由调用发起线程来执行。比如我们在一个http请求线程中调用了线程池处理异步任务，但是现在线程池满了，那么此任务将转由http请求 线程处理。缺点是会导致http请求线程阻塞，达不到异步处理的效果。优点是任务会正常执行，避免被任务执行器丢弃。
DiscardOldestPolicy
在阻塞队列最前端抛弃一个任务，然后将此任务添加到阻塞队列中排队。最前端是指最先添加到阻塞队列的任务。
DiscardPolicy
当前任务不会执行，也不会抛异常，好像什么也没有发生一样。
自定义拒绝策略
自己编写一个类，实现 RejectedExecutionHandler 接口，并重写 rejectedExecution 方法即可实现自定义拒绝策略。需要在实例化 ThreadPoolExecutor 的时候，将自定义策略配置进去。
