Java 引入 Excutor 框架将任务的提交和执行进行解耦，只需要定义好任务，然后提交给线程池即可。
使用线程池的时机：
线程池的优点：
思想：将任务提交和任务执行解耦。无需关注如何创建线程，如何调度线程来执行任务，用户只需提供实现Runnable接口的任务对象，将任务的运行逻辑提交到执行器(Executor)中，由Executor框架完成线程的调配和任务的执行部分。
线程池在内部实际上构建了一个生产者消费者模型，将线程和任务两者解耦，并不直接关联，从而良好的缓冲任务，复用线程。
线程池的运行主要分成两部分：任务管理、线程管理。
任务管理部分充当生产者的角色，当任务提交后，线程池会判断该任务后续的流转：
线程管理部分是消费者，它们被统一维护在线程池内，根据任务请求进行线程的分配，当线程执行完任务后则会继续获取新的任务去执行，最终当线程获取不到任务的时候，线程就会被回收。
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler)
//原子整型。1.声明当前线程池的状态；2.声明线程池中的线程数
//高3位是线程池的状态，低29位是线程池中的线程数量
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
private static final int COUNT_BITS = Integer.SIZE - 3; //29
private static final int CAPACITY = (1 << COUNT_BITS) - 1; //线程池中线程的最大数量2^29-1
// 高3位是线程池的状态
//-1用二进制表示就是32个1，左移29位是高3位是1，低29位是0
//111，代表线程池为RUNNING，代表正常接收任务
private static final int RUNNING = -1 << COUNT_BITS;
//000，代表线程池为SHUTDOWN状态，不接受新任务，但是内部还会处理阻塞队列中的任务，正在进行的任务也正常处理
private static final int SHUTDOWN = 0 << COUNT_BITS;
//001 代表线程池为STOP状态，不接受新任务，也不去处理阻塞队列中的任务，同时会中断正在执行的任务
private static final int STOP = 1 << COUNT_BITS;
//010 代表线程池为TIDYING状态，过渡的状态，代表线程池即将关闭
private static final int TIDYING = 2 << COUNT_BITS;
//011 TERMINATED状态，代表线程池已经关闭
private static final int TERMINATED = 3 << COUNT_BITS;
//~为取反运算符，~CAPACITY代表高3位是1，低29位是0，得到线程池的状态
private static int runStateOf(int c) { return c & ~CAPACITY; }
//获得线程池中工作线程的数量
private static int workerCountOf(int c) { return c & CAPACITY; }
//rs为高3位，表示线程池状态；wc是低29位，ctl是合并的结果
private static int ctlOf(int rs, int wc) { return rs | wc; }ctl 这个 AtomicInteger 类型，是对线程池的运行状态和线程池中有效线程的数量进行控制的一个字段， 它同时包含两部分的信息：线程池的运行状态 (runState) 和线程池内有效线程的数量 (workerCount)，高 3 位保存 runState，低29位保存 workerCount，两个变量之间互不干扰。用一个变量去存储两个值，可避免在做相关决策时，出现不一致的情况，不必为了维护两者的一致，而占用锁资源。线程池源代码中，经常出现要同时判断线程池运行状态和线程数量的情况。线程池也提供了若干方法去供用户获得线程池当前的运行状态、线程个数。这里都使用的是位运算的方式，相比于基本运算，速度也会快很多。
从前面每个状态对应的数值能得到：RUNNING < SHUTDOWN < STOP < TIDYING < TERMINATED，状态转换是由低到高的（-1 < 0 < 1 < 2 < 3）。
任务调度是线程池的主要入口，当用户提交了一个任务，接下来这个任务将如何执行都是由这个阶段决定的。这部分就是线程池的核心运行机制。
所有任务的调度都是由 execute 方法完成的，这部分完成的工作是：检查现在线程池的运行状态、运行线程数、运行策略，决定接下来执行的流程，是直接申请线程执行，或是缓冲到队列中执行，亦或是直接拒绝该任务。其执行过程如下：
执行流程：
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
//获取当前工作线程数和线程池运行状态（共32位，前3位为运行状态，后29位为运行线程数）
int c = ctl.get();
// 1. 工作线程数 < 核心线程数
if (workerCountOf(c) < corePoolSize) {
//创建核心线程执行任务，成功就返回
if (addWorker(command, true))
return;
//创建核心线程数失败，重新获取ctl
c = ctl.get();
}
// 2. 工作线程数 > 核心线程数
// 如果当前线程池状态为RUNNING，并且任务成功添加到阻塞队列
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
// 双重检查，因为从上次检查到进入此方法，线程池可能已成为SHUTDOWN状态
// 如果当前线程池状态不是 RUNNING，则从队列删除任
if (! isRunning(recheck) && remove(command))
reject(command); //拒绝策略
// 当线程池中的工作线程数为 0 时，此时阻塞队列中还有待执行的任务，核心线程数可能为 0
else if (workerCountOf(recheck) == 0)
// 新增一个线程消费阻塞队列中的任务
addWorker(null, false);
}
// 3.阻塞队列已满
// 尝试增加工作线程执行任务
else if (!addWorker(command, false))
reject(command); //拒绝策略
}任务的执行有两种可能：
线程需要从任务缓存模块中不断地取任务执行，帮助线程从阻塞队列中获取任务，实现线程管理模块和任务管理模块之间的通信。这部分策略由 getTask 方法实现。
getTask() 方法是为runWorker(Worker w)方法服务的，它的作用就是在任务队列（workQueue）中获取 task（Runnable）。
执行流程：
private Runnable getTask() {
// 表示线程是否空闲时间超时了
boolean timedOut = false;
// 自旋
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// RUNNING < SHUTDOWN < STOP < TIDYING < TERMINATED
// 1) 线程池关闭了，并且工作队列里的任务都完成了
// 2) 线程池直接进入了 STOP 或更进一步的状态，就不返回新任务
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount(); // 工作线程数-1
return null;
}
int wc = workerCountOf(c); //获取当前工作线程
// 判断当前是否需要进行超时控制，核心线程是否超时（默认false）或当前是否存在非核心线程，
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
// 如果当前线程数大于最大线程数 或 (允许超时控制 且 当前发生了空闲时间超时))
// 且当前线程数 > 1 或 阻塞队列为空
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
// 工作线程数-1
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
// 获取任务，从任务队列头部取出一个任务
// 允许超时控制：poll ，不允许超时控制：take（会阻塞）
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
// 如果获取不到任务，说明非核心线程超时了，下一轮判断确认是否退出循环。
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
线程池为了掌握线程的状态并维护线程的生命周期，设计了线程池内的工作线程 Worker。
private final class Worker extends AbstractQueuedSynchronizer implements Runnable
{
// 该worker正在运行的线程
final Thread thread;
// 将要运行的初始任务
Runnable firstTask;
// 每个线程的任务计数器
volatile long completedTasks;
// 构造方法
Worker(Runnable firstTask) {
setState(-1); // 调用runWorker()前禁止中断
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this); // 通过ThreadFactory创建一个线程
}
// 实现了Runnable接口的run方法
public void run() {
runWorker(this);
}
... // 此处省略了其他方法
}Worker 是 ThreadPoolExecutor类的内部类，实现了 Runnable 接口，并持有一个线程 thread，一个初始化的任务 firstTask 。thread 是在调用构造方法时通过 ThreadFactory 来创建的线程，可以用来执行任务；firstTask 用它来保存传入的第一个任务：：
addWorker 方法：向线程池添加一个带有任务的工作线程。
addWorker 方法有两个参数：firstTask、core。
执行流程：
1、外层循环判断线程池的状态是否可以新增工作线程。这层校验基于下面两个原则：
2、内层循环向线程池添加工作线程并返回是否添加成功的结果。
3、核心线程数量+1成功的后续操作：添加到工作线程集合，并启动工作线程
private boolean addWorker(Runnable firstTask, boolean core) {
retry: // 循环退出标志位
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// 线程池状态为RUNNANLE 或 线程池状态为SHUTDOWN但阻塞队列不为空 跳过
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())) // 换成更直观的条件语句
// (rs != SHUTDOWN || firstTask != null || workQueue.isEmpty())
// 返回false的条件就可以分解为：
//（1）线程池状态为STOP，TIDYING，TERMINATED
//（2）线程池状态为SHUTDOWN，且要执行的任务不为空
//（3）线程池状态为SHUTDOWN，且阻塞队列为空
return false;
// CAS 自旋增加线程个数
for (;;) {
int wc = workerCountOf(c); //获取工作线程个数
// 工作线程数>=线程池容量 || 工作线程数>=(核心线程数||最大线程数)
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false; // 创建工作线程失败
// 执行cas操作，线程个数 + 1
if (compareAndIncrementWorkerCount(c))
break retry; // 添加成功，退出外层循环
// 通过cas添加失败
c = ctl.get();
// 判断线程池状态是否变化，，
if (runStateOf(c) != rs)
// 变化则跳到外层循环重试，重新获取线程池状态
continue retry;
// 否则内层循环重新cas
}
}
// 简单总结上面的CAS过程：
//（1）内层循环作用是使用cas增加线程个数，如果线程个数超限则返回false，否则进行cas
//（2）cas成功则退出双循环，cas失败了，要看当前线程池的状态是否变化了
//（3）如果变了，则进入外层循环重新获取线程池状态，否则进入内层循环继续进行cas
// 走到这里说明cas成功，线程数+1，但并未被执行
boolean workerStarted = false; // 工作线程调用start()方法标志
boolean workerAdded = false; // 工作线程被添加标志
Worker w = null;
try {
w = new Worker(firstTask); //创建工作线程
final Thread t = w.thread; // 获取工作线程持有的线程实例
if (t != null) {
//获取线程池的全局锁，避免该工作线程w添加任务时，其他线程干掉了线程池，干掉线程池前需要先获取这个锁
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
int rs = runStateOf(ctl.get()); // 获取当前线程池状态
// 线程池状态为RUNNING或者（线程池状态为SHUTDOWN并且没有新任务时）
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) // 检查线程是否处于活跃状态
throw new IllegalThreadStateException();
// 线程加入到存放工作线程的HashSet容器，workers全局唯一并被mainLock持有
workers.add(w);
int s = workers.size(); //获取工作线程个数
//如果工作线程数大于之前记录的最大工作线程数，就替换
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true; //添加工作线程成功
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
t.start(); //启动工作线程 runWorker(this);
workerStarted = true; //启动工作线程成功
}
}
} finally {
if (! workerStarted) // 如果线程启动失败，则执行addWorkerFailed方法
addWorkerFailed(w);
}
//返回工作是否启动
return workerStarted;
}
runWorker(Worker w) 是线程池中真正处理任务的方法，前面的 execute() 和 addWorker() 都是在为该方法做准备和铺垫。
执行流程：
final void runWorker(Worker w) {
Thread wt = Thread.currentThread(); // 获取工作线程中用来执行任务的线程实例
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // status设置为0，允许中断
boolean completedAbruptly = true; // 线程意外终止标志
try {
//1. 任务不为空，执行任务
//2. 任务为空，通过 getTask 方法从阻塞队列中获取任务
while (task != null || (task = getTask()) != null) {
w.lock(); // 加锁，保证下方临界区代码的线程安全
// 如果线程池状态 >= STOP且当前线程还没有被中断，则主动中断线程
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt(); // 中断当前线程
try {
// 任务执行前的回调，空实现，可以在子类中自定义
beforeExecute(wt, task);
Throwable thrown = null;
try {
//开始执行任务
task.run();
} catch (RuntimeException x) {
thrown = x; throw x;
} catch (Error x) {
thrown = x; throw x;
} catch (Throwable x) {
thrown = x; throw new Error(x);
} finally {
// 任务执行后的回调，空实现，可以在子类中自定义
afterExecute(task, thrown);
}
} finally {
task = null; // 将循环变量task设置为null，表示已处理完成
w.completedTasks++; // 当前已完成的任务数+1
w.unlock();
}
}
completedAbruptly = false;
} finally {
//线程执行完毕的后续处理
processWorkerExit(w, completedAbruptly);
}
}
processWorkerExit：执行线程退出的方法
参数说明：
completedAbruptly参数为 true。执行流程：
private void processWorkerExit(Worker w, boolean completedAbruptly) {
/**
* 1.工作线程-1操作
* 1）如果completedAbruptly 为true，说明工作线程发生异常，那么将正在工作的线程数量-1
* 2）如果completedAbruptly 为false，说明工作线程无任务可以执行，由getTask()执行worker-1操作
*/
if (completedAbruptly)
decrementWorkerCount();
// 工作线程正常结束
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 将该worker已完成的任务数追加到线程池已完成的任务数
completedTaskCount += w.completedTasks;
// 从线程set集合中移除工作线程
workers.remove(w);
} finally {
mainLock.unlock();
}
// 根据线程池状态进行判断是否结束线程池
tryTerminate();
int c = ctl.get();
// 如果线程池状态是running 或 shutdown
if (runStateLessThan(c, STOP)) {
// 如果线程是突然终止的，调用addWorker()创建工作线程
// 如果当前线程不是突然终止，但当前工作线程数量小于线程池需要维护的线程数量，则创建工作线程
if (!completedAbruptly) {
// min为线程池需要维护的线程数量，
// 允许核心线程超时销毁，min = 0，否则为核心线程数
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
// 如果min == 0且阻塞队列不为空
if (min == 0 && ! workQueue.isEmpty())
min = 1;
// 当前工作线程数>=线程池需要维护的线程数量，不需要创建新的线程
if (workerCountOf(c) >= min)
return; // replacement not needed
}
addWorker(null, false);
}
}方式一：通过 ThreadPoolExecutor 构造函数实现（推荐）
public class ThreadPoolExecutorDemo {
private static final int CORE_POOL_SIZE = 5;
private static final int MAX_POOL_SIZE = 10;
private static final int QUEUE_CAPACITY = 100;
private static final Long KEEP_ALIVE_TIME = 1L;
public static void main(String[] args) {
//使用阿里巴巴推荐的创建线程池的方式
//通过ThreadPoolExecutor构造函数自定义参数创建
ThreadPoolExecutor executor = new ThreadPoolExecutor(
CORE_POOL_SIZE,
MAX_POOL_SIZE,
KEEP_ALIVE_TIME,
TimeUnit.SECONDS,
new ArrayBlockingQueue<>(QUEUE_CAPACITY),
new ThreadPoolExecutor.CallerRunsPolicy());
for (int i = 0; i < 10; i++) {
//创建WorkerThread对象（WorkerThread类实现了Runnable 接口）
Runnable worker = new MyRunnable("" + i);
//执行Runnable
executor.execute(worker);
}
//终止线程池
executor.shutdown();
while (!executor.isTerminated()) {
}
System.out.println("Finished all threads");
}
}
方式二：通过 Executor 框架的工具类 Executors 来实现我们可以创建三种类型的 ThreadPoolExecutor ：
newFixedThreadPool
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
// 使用的是默认工厂和拒绝策略。
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), defaultHandler);
}特点：
适用于任务量已知，相对耗时的任务
使用的是默认工厂，可以看到创建的是非守护线程
static class DefaultThreadFactory implements ThreadFactory {
private static final AtomicInteger poolNumber = new AtomicInteger(1);
private final ThreadGroup group;
private final AtomicInteger threadNumber = new AtomicInteger(1);
private final String namePrefix;
DefaultThreadFactory() {
SecurityManager s = System.getSecurityManager();
group = (s != null) ? s.getThreadGroup() :
Thread.currentThread().getThreadGroup();
namePrefix = "pool-" +
poolNumber.getAndIncrement() +
"-thread-";
}
public Thread newThread(Runnable r) {
Thread t = new Thread(group, r,
namePrefix + threadNumber.getAndIncrement(),
0);
if (t.isDaemon())
t.setDaemon(false);
if (t.getPriority() != Thread.NORM_PRIORITY)
t.setPriority(Thread.NORM_PRIORITY);
return t;
}
}我们可以自定义工厂，重写newThread方法
ExecutorService pool = Executors.newFixedThreadPool(2, new ThreadFactory() {
private AtomicInteger t = new AtomicInteger(1);
@Override
public Thread newThread(Runnable r) {
return new Thread(r,"mypool-" + t.getAndIncrement());
}
});输出
09:22:50.927 [mypool-1] DEBUG p8_1.testThreadPool.Test1 - 1
09:22:50.927 [mypool-2] DEBUG p8_1.testThreadPool.Test1 - 2
09:22:50.931 [mypool-1] DEBUG p8_1.testThreadPool.Test1 - 3
newCachedThreadPool
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}特点：
SynchronousQueue<Integer> integers = new SynchronousQueue<>();
new Thread(() -> {
try {
log.debug("putting {} ", 1);
integers.put(1);
log.debug("{} putted...", 1);
log.debug("putting {} ", 2);
integers.put(2);
log.debug("{} putted...", 2);
} catch (InterruptedException e) {
e.printStackTrace();
}
},"t1").start();
TimeUnit.SECONDS.sleep(1);
new Thread(() -> {
try {
log.debug("taking {}", integers.take());
} catch (InterruptedException e) {
e.printStackTrace();
}
},"t2").start();
TimeUnit.SECONDS.sleep(1);
new Thread(() -> {
try {
log.debug("taking {}", integers.take());
} catch (InterruptedException e) {
e.printStackTrace();
}
},"t3").start();
}输出
09:32:53.238 [t1] DEBUG p8_1.testThreadPool.Test2 - putting 1
09:32:54.237 [t1] DEBUG p8_1.testThreadPool.Test2 - 1 putted...
09:32:54.237 [t2] DEBUG p8_1.testThreadPool.Test2 - taking 1
09:32:54.238 [t1] DEBUG p8_1.testThreadPool.Test2 - putting 2
09:32:55.244 [t1] DEBUG p8_1.testThreadPool.Test2 - 2 putted...
09:32:55.244 [t3] DEBUG p8_1.testThreadPool.Test2 - taking 2整个线程池表现为线程数会根据任务量不断增长，没有上限，当任务执行完毕，空闲 1分钟后释放线程。 适合任务数比较密集，但每个任务执行时间较短的情况
newSingleThreadExecutor
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}使用场景：
希望多个任务排队执行。线程数固定为 1，任务数多于 1 时，会放入无界队列排队。任务执行完毕，这唯一的线程也不会被释放。
区别：
// 强转为ThreadPoolExecutor
ThreadPoolExecutor threadPool = (ThreadPoolExecutor) Executors.newFixedThreadPool(1);
// 改变核心线程数
threadPool.setCorePoolSize(2)
装饰者模式具体体现：内部调用了 ThreadPoolExecutor 的构造方法，传入的 corePoolSize 和maximumPoolSize 都为1，然后将该对象传给了 FinalizableDelegatedExecutorService，该类修饰了 ThreadPoolExecutor，让外部无法调用 ThreadPoolExecutor 内部的某些方法来修改所创建的线程池的大小。
public class Executors {
static class FinalizableDelegatedExecutorService
extends DelegatedExecutorService {
FinalizableDelegatedExecutorService(ExecutorService executor) {
super(executor);
}
protected void finalize() {
super.shutdown();
}
}
static class DelegatedExecutorService extends AbstractExecutorService {
private final ExecutorService e;
DelegatedExecutorService(ExecutorService executor) { e = executor; }
public void execute(Runnable command) { e.execute(command); }
public void shutdown() { e.shutdown(); }
public List<Runnable> shutdownNow() { return e.shutdownNow(); }
public boolean isShutdown() { return e.isShutdown(); }
public boolean isTerminated() { return e.isTerminated(); }
public boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException {
return e.awaitTermination(timeout, unit);
}
public Future<?> submit(Runnable task) {
return e.submit(task);
}
public <T> Future<T> submit(Callable<T> task) {
return e.submit(task);
}
public <T> Future<T> submit(Runnable task, T result) {
return e.submit(task, result);
}
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
throws InterruptedException {
return e.invokeAll(tasks);
}
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException {
return e.invokeAll(tasks, timeout, unit);
}
public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
throws InterruptedException, ExecutionException {
return e.invokeAny(tasks);
}
public <T> T invokeAny(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {
return e.invokeAny(tasks, timeout, unit);
}
}
}测试：任务执行失败，线程池重新创建了一个新的线程
ExecutorService pool = Executors.newSingleThreadExecutor();
pool.execute(() -> {
log.debug("{}",1 / 0);
});
pool.execute(() -> {
log.debug("1");
});
pool.execute(() -> {
log.debug("2");
});Exception in thread "pool-1-thread-1" java.lang.ArithmeticException: / by zero
at p8_1.testThreadPool.Test3.lambda$main$0(Test3.java:17)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:748)
10:29:45.500 [pool-1-thread-2] DEBUG p8_1.testThreadPool.Test3 - 1
10:29:45.503 [pool-1-thread-2] DEBUG p8_1.testThreadPool.Test3 - 2
// 执行任务
void execute(Runnable command);
// 提交任务 task，用返回值 Future 获得任务执行结果
<T> Future<T> submit(Callable<T> task);
// 提交 tasks 中所有任务
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
throws InterruptedException;
// 提交 tasks 中所有任务，带超时时间
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException;
// 提交 tasks 中所有任务，哪个任务先成功执行完毕，返回此任务执行结果，其它任务取消
<T> T invokeAny(Collection<? extends Callable<T>> tasks)
throws InterruptedException, ExecutionException;
// 提交 tasks 中所有任务，哪个任务先成功执行完毕，返回此任务执行结果，其它任务取消，带超时时间
<T> T invokeAny(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException;测试submit
ExecutorService pool = Executors.newFixedThreadPool(2);
Future<String> future = pool.submit(() -> {
log.debug("running");
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "ok";
});
log.debug("{}",future.get());10:33:20.269 [pool-1-thread-1] DEBUG p8_1.testThreadPool.Test4 - running
10:33:21.284 [main] DEBUG p8_1.testThreadPool.Test4 - ok测试invokeAll
ExecutorService pool = Executors.newFixedThreadPool(2);
List<Future<String>> futures = pool.invokeAll(Arrays.asList(
() -> {
log.debug("beging");
TimeUnit.SECONDS.sleep(1);
return "1";
},
() -> {
log.debug("beging");
TimeUnit.SECONDS.sleep(2);
return "2";
},
() -> {
log.debug("beging");
TimeUnit.SECONDS.sleep(3);
return "3";
}
));
futures.forEach( future -> {
try {
log.debug("{}",future.get());
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
});10:39:01.332 [pool-1-thread-2] DEBUG p8_1.testThreadPool.Test5 - beging
10:39:01.332 [pool-1-thread-1] DEBUG p8_1.testThreadPool.Test5 - beging
10:39:02.337 [pool-1-thread-1] DEBUG p8_1.testThreadPool.Test5 - beging
10:39:05.341 [main] DEBUG p8_1.testThreadPool.Test5 - 1
10:39:05.343 [main] DEBUG p8_1.testThreadPool.Test5 - 2
10:39:05.343 [main] DEBUG p8_1.testThreadPool.Test5 - 3测试invokeAny
ExecutorService pool = Executors.newFixedThreadPool(2);
String result = pool.invokeAny(Arrays.asList(
() -> {
log.debug("beging 1");
TimeUnit.SECONDS.sleep(2);
log.debug("end 1");
return "1";
},
() -> {
log.debug("beging 2");
TimeUnit.SECONDS.sleep(1);
log.debug("end 2");
return "2";
},
() -> {
log.debug("beging 3");
TimeUnit.SECONDS.sleep(3);
log.debug("end 3");
return "3";
}
));
log.debug("{}",result);
10:40:12.286 [pool-1-thread-1] DEBUG p8_1.testThreadPool.Test6 - beging 1
10:40:12.286 [pool-1-thread-2] DEBUG p8_1.testThreadPool.Test6 - beging 2
10:40:13.289 [pool-1-thread-2] DEBUG p8_1.testThreadPool.Test6 - end 2
10:40:13.289 [pool-1-thread-2] DEBUG p8_1.testThreadPool.Test6 - beging 3
10:40:13.289 [main] DEBUG p8_1.testThreadPool.Test6 - 2线程池状态变为 SHUTDOWN
void shutdown();测试
ExecutorService pool = Executors.newFixedThreadPool(2);
pool.submit(() -> {
log.debug("task 1 running");
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
log.debug("task 1 finish");
});
pool.submit(() -> {
log.debug("task 2 running");
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
log.debug("task 2 finish");
});
pool.submit(() -> {
log.debug("task 3 running");
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
log.debug("task 3 finish");
});
log.debug("shutdown");
pool.shutdown();
pool.submit(() -> {
log.debug("task 4 running");
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
log.debug("task 4 finish");
});10:58:11.923 [main] DEBUG c.Test - shutdown
10:58:11.923 [pool-1-thread-2] DEBUG c.Test - task 2 running
10:58:11.923 [pool-1-thread-1] DEBUG c.Test - task 1 running
Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task java.util.concurrent.FutureTask@1376c05c rejected from java.util.concurrent.ThreadPoolExecutor@51521cc1[Shutting down, pool size = 2, active threads = 2, queued tasks = 1, completed tasks = 0]
at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2047)
at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:823)
at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1369)
at java.util.concurrent.AbstractExecutorService.submit(AbstractExecutorService.java:112)
at p8_1.testThreadPool.Test7.main(Test7.java:53)
10:58:12.938 [pool-1-thread-1] DEBUG c.Test - task 1 finish
10:58:12.938 [pool-1-thread-2] DEBUG c.Test - task 2 finish
10:58:12.938 [pool-1-thread-1] DEBUG c.Test - task 3 running
10:58:13.952 [pool-1-thread-1] DEBUG c.Test - task 3 finish源码
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
// 修改线程池状态
advanceRunState(SHUTDOWN);
// 仅会打断空闲线程
interruptIdleWorkers();
onShutdown(); // 扩展点 ScheduledThreadPoolExecutor
} finally {
mainLock.unlock();
}
// 尝试终结(没有运行的线程可以立刻终结，如果还有运行的线程也不会等)
tryTerminate();
}
线程池状态变为 STOP
List<Runnable> shutdownNow();测试
ExecutorService pool = Executors.newFixedThreadPool(2);
pool.submit(() -> {
log.debug("task 1 running");
try {
TimeUnit.SECONDS.sleep(1);
log.debug("task 1 finish");
} catch (InterruptedException e) {
e.printStackTrace();
}
return "1";
});
pool.submit(() -> {
log.debug("task 2 running");
try {
TimeUnit.SECONDS.sleep(1);
log.debug("task 2 finish");
} catch (InterruptedException e) {
e.printStackTrace();
}
return "2";
});
pool.submit(() -> {
log.debug("task 3 running");
try {
TimeUnit.SECONDS.sleep(1);
log.debug("task 3 finish");
} catch (InterruptedException e) {
e.printStackTrace();
}
return "3";
});
log.debug("shutdownNow");
List<Runnable> runnables = pool.shutdownNow();
log.debug("other....{}",runnables.size());10:53:02.880 [main] DEBUG c.Test8 - shutdownNow
10:53:02.880 [pool-1-thread-1] DEBUG c.Test8 - task 1 running
10:53:02.880 [pool-1-thread-2] DEBUG c.Test8 - task 2 running
java.lang.InterruptedException: sleep interrupted
at java.lang.Thread.sleep(Native Method)
at java.lang.Thread.sleep(Thread.java:340)
at java.util.concurrent.TimeUnit.sleep(TimeUnit.java:386)
at p8_1.testThreadPool.Test8.lambda$main$1(Test8.java:31)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:748)
java.lang.InterruptedException: sleep interrupted
at java.lang.Thread.sleep(Native Method)
at java.lang.Thread.sleep(Thread.java:340)
at java.util.concurrent.TimeUnit.sleep(TimeUnit.java:386)
at p8_1.testThreadPool.Test8.lambda$main$0(Test8.java:21)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:748)
10:53:02.883 [main] DEBUG c.Test8 - other....1源码
public List<Runnable> shutdownNow() {
List<Runnable> tasks;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
// 修改线程池状态
advanceRunState(STOP);
// 打断所有线程
interruptWorkers();
// 获取队列中剩余任务
tasks = drainQueue();
} finally {
mainLock.unlock();
}
// 尝试终结
tryTerminate();
return tasks;
}// 不在 RUNNING 状态的线程池，此方法就返回 true
boolean isShutdown();
// 线程池状态是否是 TERMINATED
boolean isTerminated();
// 调用 shutdown 后，由于调用线程并不会等待所有任务运行结束，
// 因此如果它想在线程池 TERMINATED 后做些事情，可以利用此方法等待
boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException;测试awaitTermination
ExecutorService pool = Executors.newFixedThreadPool(2);
pool.submit(() -> {
log.debug("task 1 running");
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
log.debug("task 1 finish");
});
pool.submit(() -> {
log.debug("task 2 running");
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
log.debug("task 2 finish");
});
pool.submit(() -> {
log.debug("task 3 running");
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
log.debug("task 3 finish");
});
log.debug("shutdown");
pool.shutdown();
pool.awaitTermination(3,TimeUnit.SECONDS);
log.debug("main-Thread other thing");
11:00:32.136 [main] DEBUG c.Test - shutdown
11:00:32.136 [pool-1-thread-1] DEBUG c.Test - task 1 running
11:00:32.136 [pool-1-thread-2] DEBUG c.Test - task 2 running
11:00:33.140 [pool-1-thread-1] DEBUG c.Test - task 1 finish
11:00:33.140 [pool-1-thread-2] DEBUG c.Test - task 2 finish
11:00:33.140 [pool-1-thread-2] DEBUG c.Test - task 3 running
11:00:34.151 [pool-1-thread-2] DEBUG c.Test - task 3 finish
11:00:34.151 [main] DEBUG c.Test - main-Thread other thing
在『任务调度线程池』功能加入之前，可以使用 java.util.Timer 来实现定时功能，Timer 的优点在于简单易用，但由于所有任务都是由同一个线程来调度，因此所有任务都是串行执行的，同一时间只能有一个任务在执行，前一个任务的延迟或异常都将会影响到之后的任务。
Timer timer = new Timer();
TimerTask task1 = new TimerTask() {
@Override
public void run() {
log.debug("1");
try {
TimeUnit.SECONDS.sleep(2);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
};
TimerTask task2 = new TimerTask() {
@Override
public void run() {
log.debug("2");
}
};
log.debug("start");
timer.schedule(task1,1000);
timer.schedule(task2,1000);注意观察时间
11:07:15.238 [main] DEBUG c.Test9 - start
11:07:16.251 [Timer-0] DEBUG c.Test9 - 1
11:07:18.266 [Timer-0] DEBUG c.Test9 - 2使用 ScheduledExecutorService 改写
ScheduledExecutorService pool = Executors.newScheduledThreadPool(2);
pool.schedule(() -> {
System.out.println("任务1，执行时间：" + new Date());
try {
TimeUnit.SECONDS.sleep(2);
} catch (InterruptedException e) {
e.printStackTrace();
}
}, 1, TimeUnit.SECONDS);
pool.schedule(() -> {
System.out.println("任务2，执行时间：" + new Date());
}, 2, TimeUnit.SECONDS);任务1，执行时间：Mon Oct 17 11:09:10 CST 2022
任务2，执行时间：Mon Oct 17 11:09:11 CST 2022
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
long initialDelay,
long period,
TimeUnit unit);任务执行时间超过了间隔时间
ScheduledExecutorService pool = Executors.newScheduledThreadPool(1);
log.debug("start...");
pool.scheduleAtFixedRate(() -> {
log.debug("running...");
try {
TimeUnit.SECONDS.sleep(2);
} catch (InterruptedException e) {
e.printStackTrace();
}
}, 1, 1, TimeUnit.SECONDS);输出分析：一开始，延时 1s，接下来，由于任务执行时间 > 间隔时间，间隔被『撑』到了 2s
11:13:15.104 [main] DEBUG c.Test11 - start...
11:13:16.160 [pool-1-thread-1] DEBUG c.Test11 - running...
11:13:18.172 [pool-1-thread-1] DEBUG c.Test11 - running...
11:13:20.184 [pool-1-thread-1] DEBUG c.Test11 - running... public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
long initialDelay,
long delay,
TimeUnit unit);测试
ScheduledExecutorService pool = Executors.newScheduledThreadPool(1);
log.debug("start...");
pool.scheduleWithFixedDelay(() -> {
log.debug("running...");
try {
TimeUnit.SECONDS.sleep(2);
} catch (InterruptedException e) {
e.printStackTrace();
}
},1,1,TimeUnit.SECONDS);输出分析：一开始，延时 1s，scheduleWithFixedDelay 的间隔是上一个任务结束 <-> 延时 <-> 下一个任务开始，所以间隔都是 3s
11:14:44.395 [main] DEBUG c.Test11 - start...
11:14:45.455 [pool-1-thread-1] DEBUG c.Test11 - running...
11:14:48.474 [pool-1-thread-1] DEBUG c.Test11 - running...
11:14:51.486 [pool-1-thread-1] DEBUG c.Test11 - running...整个线程池表现为：线程数固定，任务数多于线程数时，会放入无界队列排队。任务执行完毕，这些线程也不会被释放。用来执行延迟或反复执行的任务
// 获得当前时间
LocalDateTime now = LocalDateTime.now();
// 获取周四时间
LocalDateTime thursday =
now.with(DayOfWeek.THURSDAY).withHour(18).withMinute(0).withSecond(0).withNano(0);
// 如果当前时间已经超过本周四 18:00:00.000，那么找下周四 18:00:00.000
if(now.compareTo(thursday) >= 0){
thursday = thursday.plusWeeks(1);
}
// 计算时间差，即延时执行时间
long initailDelay = Duration.between(now, thursday).toMillis();
// 计算间隔时间，即 1 周的毫秒值
long period = 7 * 24 * 3600 * 1000;
ScheduledExecutorService pool = Executors.newScheduledThreadPool(1);
pool.scheduleAtFixedRate(()->{
System.out.println("running");
}, initailDelay, period, TimeUnit.MILLISECONDS);
ExecutorService pool = Executors.newFixedThreadPool(1);
pool.execute(() -> {
try {
log.debug("task1");
int i = 1 / 0;
} catch (Exception e) {
log.error("error:", e);
}
});11:21:59.668 [pool-1-thread-1] DEBUG c.TestException - task1
11:21:59.673 [pool-1-thread-1] ERROR c.TestException - error:
java.lang.ArithmeticException: / by zero
at p8_1.testThreadPool.TestException.lambda$main$0(TestException.java:22)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:748) ExecutorService pool = Executors.newFixedThreadPool(1);
Future<Boolean> f = pool.submit(() -> {
log.debug("task1");
int i = 1 / 0;
return true;
});
log.debug("result:{}", f.get());11:19:56.760 [pool-1-thread-1] DEBUG c.TestException - task1
Exception in thread "main" java.util.concurrent.ExecutionException: java.lang.ArithmeticException: / by zero
at java.util.concurrent.FutureTask.report(FutureTask.java:122)
at java.util.concurrent.FutureTask.get(FutureTask.java:192)
at p8_1.testThreadPool.TestException.main(TestException.java:33)
Caused by: java.lang.ArithmeticException: / by zero
at p8_1.testThreadPool.TestException.lambda$main$0(TestException.java:30)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:748)
CPU 密集型运算
通常采用 cpu 核数 + 1 能够实现最优的 CPU 利用率，+1 是保证当线程由于页缺失故障（操作系统）或其它原因导致暂停时，额外的这个线程就能顶上去，保证 CPU 时钟周期不被浪费。
I/O 密集型运算
CPU 不总是处于繁忙状态，例如，当你执行业务计算时，这时候会使用 CPU 资源，但当你执行 I/O 操作时、远程RPC 调用时，包括进行数据库操作时，这时候 CPU 就闲下来了，你可以利用多线程提高它的利用率。
经验公式如下：
例如 4 核 CPU 计算时间是 50%，其它等待时间是 50%，期望 cpu 被 100% 利用，套用公式
参考文章：线程池源码解析_bullshitter的博客-CSDN博客_线程池源码解析