剑指JUC原理-13.线程池（上）：https://developer.aliyun.com/article/1413640
FutureTask的run方法：
public void run() { /*compareAndSwapObject(this, runnerOffset,]null, Thread.currentThread())) 其中第一个参数为需要改变的对象，第二个为偏移量，第三个参数为期待的值，第四个为更新后的值。 */ if (state != NEW || !UNSAFE.compareAndSwapObject(this, runnerOffset, null, Thread.currentThread())) return; try { Callable<V> c = callable; if (c != null && state == NEW) { V result; boolean ran; try { //call()方法是由FutureTask调用的,说明call()不是异步执行的 result = c.call(); ran = true; } catch (Throwable ex) { result = null; ran = false; //设置异常 setException(ex); } if (ran) set(result); } } finally { // runner must be non-null until state is settled to // prevent concurrent calls to run() runner = null; // state must be re-read after nulling runner to prevent // leaked interrupts int s = state; //判断是否被中断 if (s >= INTERRUPTING) handlePossibleCancellationInterrupt(s); } }
set方法：
protected void set(V v) { // NEW -> COMPLETING if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) { //返回结果,也包括异常 outcome = v; //COMPLETING -> NORMAL UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state //唤醒等待的线程 finishCompletion(); } }
将返回值和状态赋值回去。
report方法：
private V report(int s) throws ExecutionException { Object x = outcome; if (s == NORMAL) return (V)x; if (s >= CANCELLED) throw new CancellationException(); throw new ExecutionException((Throwable)x); }
get方法：
public V get() throws InterruptedException, ExecutionException { int s = state; //是否是未完成状态,是则等待 if (s <= COMPLETING) //等待过程 s = awaitDone(false, 0L); return report(s); } /** * @throws CancellationException {@inheritDoc} */ public V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { if (unit == null) throw new NullPointerException(); int s = state; if (s <= COMPLETING && (s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING) throw new TimeoutException(); return report(s); }
run方法解析完成，使用get方法获取返回结果。
总结：Future只实现了异步，而没有实现回调，主线程get时会阻塞，可以轮询以便获取异步调用是否完成。
关闭线程池
shutdown流程
- 修改线程池状态为
SHUTDOWN - 不再接收新提交的任务
- 中断线程池中空闲的线程
- 第③步只是中断了空闲的线程，但正在执行的任务以及线程池任务队列中的任务会继续执行完毕
shutdownNow流程
- 修改线程池状态为
STOP - 不再接收任务提交
- 尝试中断线程池中所有的线程（包括正在执行的线程）
- 返回正在等待执行的任务列表
List
此时线程池中等待队列中的任务不会被执行，正在执行的任务也可能被终止（为什么是可能呢？因为如果正常执行的任务如果不响应中断，那么就不会被终止，直到任务执行完毕）
问题说明
线程是如何中断的
中断线程池中的线程的方法是通过调用 Thread.interrupt()方法来实现的，这种方法的作用有限，如果线程中没有sleep 、wait、Condition、定时锁等应用, interrupt() 方法是无法中断当前的线程的（因为sleep、condition、await这些是响应中断的）。所以，shutdownNow()并不代表线程池就一定立即就能退出，它也可能必须要等待所有正在执行的任务都执行完成了才能退出。但是大多数时候是能立即退出的。
为什么修改线程池状态为shutdown以后线程池就不能接收新任务了
在向线程池提交任务的时候，会先检查线程池状态， 线程池状态为非关闭（或停止）时才能提交任务，这里已经将线程池状态修改为shutdown了，自然就不能接受新的任务提交了，可参考execute(Runnable command)逻辑和 addWorker逻辑）
execute方法
public void execute(Runnable command) { if (command == null) throw new NullPointerException(); int c = ctl.get(); // 【 1 】、worker数量比核心线程数小，直接创建worker执行任务 if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) return; c = ctl.get(); } // 【 2 】、worker数量超过核心线程数，任务直接进入队列。这里进入队列前先判断了线程池状态 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 】、 如果线程池不是运行状态，或者任务进入队列失败，则尝试创建worker执行任务（即：线程阻塞队列满了但线程池中的线程数没达到最大线程数， // 则新开启一个线程去执行该任务）。 // 这儿有3点需要注意： // 1. 线程池不是运行状态时，addWorker内部会判断线程池状态 // 2. addWorker第2个参数表示是否创建核心线程 // 3. addWorker返回false，则说明任务执行失败，需要执行reject操作 else if (!addWorker(command, false)) reject(command); }
addWorker方法
// 新建一个worker w = new Worker(firstTask); int rs = runStateOf(ctl.get()); // 这儿需要重新检查线程池状态，即线程池状态为shutdown以后不能再提交任务了 if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) { // 添加到工作线程集合中 workers.add(w); int s = workers.size(); // 更新largestPoolSize变量的值 if (s > largestPoolSize) largestPoolSize = s; workerAdded = true; }
源码比较
shutdown源码
public void shutdown() { final ReentrantLock mainLock = this.mainLock; // 获取线程池的锁 mainLock.lock(); try { // 检查关闭进入许可 checkShutdownAccess(); // 将线程池状态改为SHUTDOWN advanceRunState(SHUTDOWN); // 【中断线程池中空闲线程】，注意和下面的shutdownNow方法中的进行对比 interruptIdleWorkers(); // 留给定时任务线程池的钩子方法，这里没有实现，在定时任务线程池中有实现 onShutdown(); // hook for ScheduledThreadPoolExecutor } finally { mainLock.unlock(); } // ①、如果线程池状态为正在运行 或 已经是 TIDYING 状态以上了 或者 线程池状态为shutdown但是等待队列中还有任务， // 那么这个方法什么都不做，直接返回 // ②、如何线程池中还有未被中断的线程，则这里会再次去中断他（并且利用中断传播 从等待队列中删除等待的worker） // ③、如果线程池的状态是shutdown，并且等待队列中已经没有任务了，那么此时会把线程池状态转换为 TIDYING， // 并唤醒所有调用awaitTermination（）等待线程池关闭的线程 tryTerminate(); }
shutdownNow源码
public List<Runnable> shutdownNow() { List<Runnable> tasks; // 获取线程池的锁 final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { // 检查关闭进入许可 checkShutdownAccess(); // 将线程池状态改为STOP advanceRunState(STOP); // 【中断所有线程】 interruptWorkers(); // 获取任务队列里未完成任务 tasks = drainQueue(); } finally { mainLock.unlock(); } // ①、如果线程池状态为正在运行 或 已经是 TIDYING 状态以上了 或者 线程池状态为shutdown但是等待队列中还有任务， // 那么这个方法什么都不做，直接返回 // ②、如何线程池中还有未被中断的线程，则这里会再次去中断他（并且利用中断传播 从等待队列中删除等待的worker） // ③、如果线程池的状态是shutdown，并且等待队列中已经没有任务了，那么此时会把线程池状态转换为 TIDYING， // 并唤醒所有调用awaitTermination（）等待线程池关闭的线程 tryTerminate(); return tasks; }
通过比较shutdown源码和shutdownNow的源码我们可以发现，这两个方法最大的不同在于中断线程的地方：
- 首先，需要再次明确的一点是，中断线程并不是立即把这个线程停止，而是把这个线程的【中断状态】设置为true，表示有其他线程来中断过这个线程。
- shutdown方法调用interruptIdleWorkers()方法中断的只是线程池中空闲的线程，那么也就说明线程池中正在工作的线程没有被中断，所以说正在工作的线程会继续执行完毕，并且正在工作的线程也会去任务队列中将已经提交的任务取出来并执行。
- shutdownNow方法调用的是interruptWorkers（）方法，该方法会逐一遍历线程池中的每一个线程（包括空闲线程和正在工作的线程）并且去中断他，所以说当调用shutdownNow方法的时候，所有线程都会被中断，等待队列中的任务不会被执行，正在执行的任务也可能会中断退出（为什么是可能而不是一定？因为如果正在执行的线程不响应中断，那么他就会继续运行）
异步模式之工作线程
定义
让有限的工作线程（Worker Thread）来轮流异步处理无限多的任务。也可以将其归类为分工模式，它的典型实现就是线程池，也体现了经典设计模式中的享元模式。
例如，海底捞的服务员（线程），轮流处理每位客人的点餐（任务），如果为每位客人都配一名专属的服务员，那么成本就太高了
注意，不同任务类型应该使用不同的线程池，这样能够避免饥饿，并能提升效率
例如，如果一个餐馆的工人既要招呼客人（任务类型A），又要到后厨做菜（任务类型B）显然效率不咋地，分成服务员（线程池A）与厨师（线程池B）更为合理，当然你能想到更细致的分工
饥饿
固定大小线程池会有饥饿现象
- 两个工人是同一个线程池中的两个线程
- 他们要做的事情是：为客人点餐和到后厨做菜，这是两个阶段的工作。客人点餐：必须先点完餐，等菜做好，上菜，在此期间处理点餐的工人必须等待、后厨做菜：没啥说的，做就是了
- 比如工人A 处理了点餐任务，接下来它要等着 工人B 把菜做好，然后上菜，他俩也配合的蛮好
- 但现在同时来了两个客人，这个时候工人A 和工人B 都去处理点餐了，这时没人做饭了，饥饿
public class TestDeadLock { static final List<String> MENU = Arrays.asList("地三鲜", "宫保鸡丁", "辣子鸡丁", "烤鸡翅"); static Random RANDOM = new Random(); static String cooking() { return MENU.get(RANDOM.nextInt(MENU.size())); } public static void main(String[] args) { ExecutorService executorService = Executors.newFixedThreadPool(2); executorService.execute(() -> { log.debug("处理点餐..."); Future<String> f = executorService.submit(() -> { log.debug("做菜"); return cooking(); }); try { log.debug("上菜: {}", f.get()); } catch (InterruptedException | ExecutionException e) { e.printStackTrace(); } }); // executorService.execute(() -> { // log.debug("处理点餐..."); // Future<String> f = executorService.submit(() -> { // log.debug("做菜"); // return cooking(); // }); // try { // log.debug("上菜: {}", f.get()); // } catch (InterruptedException | ExecutionException e) { // e.printStackTrace(); // } // }); } }
输出
17:21:27.883 c.TestDeadLock [pool-1-thread-1] - 处理点餐... 17:21:27.891 c.TestDeadLock [pool-1-thread-2] - 做菜 17:21:27.891 c.TestDeadLock [pool-1-thread-1] - 上菜: 烤鸡翅
当注释取消后，可能的输出
17:08:41.339 c.TestDeadLock [pool-1-thread-2] - 处理点餐... 17:08:41.339 c.TestDeadLock [pool-1-thread-1] - 处理点餐...
解决方法可以增加线程池的大小，不过不是根本解决方案，还是前面提到的，不同的任务类型，采用不同的线程池，例如：
public class TestDeadLock { static final List<String> MENU = Arrays.asList("地三鲜", "宫保鸡丁", "辣子鸡丁", "烤鸡翅"); static Random RANDOM = new Random(); static String cooking() { return MENU.get(RANDOM.nextInt(MENU.size())); } public static void main(String[] args) { ExecutorService waiterPool = Executors.newFixedThreadPool(1); ExecutorService cookPool = Executors.newFixedThreadPool(1); waiterPool.execute(() -> { log.debug("处理点餐..."); Future<String> f = cookPool.submit(() -> { log.debug("做菜"); return cooking(); }); try { log.debug("上菜: {}", f.get()); } catch (InterruptedException | ExecutionException e) { e.printStackTrace(); } }); waiterPool.execute(() -> { log.debug("处理点餐..."); Future<String> f = cookPool.submit(() -> { log.debug("做菜"); return cooking(); }); try { log.debug("上菜: {}", f.get()); } catch (InterruptedException | ExecutionException e) { e.printStackTrace(); } }); } }
输出
17:25:14.626 c.TestDeadLock [pool-1-thread-1] - 处理点餐... 17:25:14.630 c.TestDeadLock [pool-2-thread-1] - 做菜 17:25:14.631 c.TestDeadLock [pool-1-thread-1] - 上菜: 地三鲜 17:25:14.632 c.TestDeadLock [pool-1-thread-1] - 处理点餐... 17:25:14.632 c.TestDeadLock [pool-2-thread-1] - 做菜 17:25:14.632 c.TestDeadLock [pool-1-thread-1] - 上菜: 辣子鸡丁
创建多少线程池合适
- 过小会导致程序不能充分地利用系统资源、容易导致饥饿
- 过大会导致更多的线程上下文切换，占用更多内存
CPU 密集型运算
通常采用 cpu 核数 + 1 能够实现最优的 CPU 利用率，+1 是保证当线程由于页缺失故障（操作系统）或其它原因导致暂停时，额外的这个线程就能顶上去，保证 CPU 时钟周期不被浪费
I/O 密集型运算
CPU 不总是处于繁忙状态，例如，当你执行业务计算时，这时候会使用 CPU 资源，但当你执行 I/O 操作时、远程RPC 调用时，包括进行数据库操作时，这时候 CPU 就闲下来了，你可以利用多线程提高它的利用率。
经验公式如下
线程数 = 核数 * 期望 CPU 利用率 * 总时间(CPU计算时间+等待时间) / CPU 计算时间
例如 4 核 CPU 计算时间是 50% ，其它等待时间是 50%，期望 cpu 被 100% 利用，套用公式
4 * 100% * 100% / 50% = 8
例如 4 核 CPU 计算时间是 10% ，其它等待时间是 90%，期望 cpu 被 100% 利用，套用公式
4 * 100% * 100% / 10% = 40
任务调度线程池
在『任务调度线程池』功能加入之前，可以使用 java.util.Timer 来实现定时功能，Timer 的优点在于简单易用，但由于所有任务都是由同一个线程来调度，因此所有任务都是串行执行的，同一时间只能有一个任务在执行，前一个任务的延迟或异常都将会影响到之后的任务。
public static void main(String[] args) { Timer timer = new Timer(); TimerTask task1 = new TimerTask() { @Override public void run() { log.debug("task 1"); sleep(2); } }; TimerTask task2 = new TimerTask() { @Override public void run() { log.debug("task 2"); } }; // 使用 timer 添加两个任务，希望它们都在 1s 后执行 // 但由于 timer 内只有一个线程来顺序执行队列中的任务，因此『任务1』的延时，影响了『任务2』的执行 timer.schedule(task1, 1000); timer.schedule(task2, 1000); }
输出
20:46:09.444 c.TestTimer [main] - start... 20:46:10.447 c.TestTimer [Timer-0] - task 1 20:46:12.448 c.TestTimer [Timer-0] - task 2
使用 ScheduledExecutorService 改写：
ScheduledExecutorService executor = Executors.newScheduledThreadPool(2); // 添加两个任务，希望它们都在 1s 后执行 executor.schedule(() -> { System.out.println("任务1，执行时间：" + new Date()); try { Thread.sleep(2000); } catch (InterruptedException e) { } }, 1000, TimeUnit.MILLISECONDS); executor.schedule(() -> { System.out.println("任务2，执行时间：" + new Date()); }, 1000, TimeUnit.MILLISECONDS);
输出
任务1，执行时间：Thu Jan 03 12:45:17 CST 2019 任务2，执行时间：Thu Jan 03 12:45:17 CST 2019
异常的情况
ScheduledExecutorService executor = Executors.newScheduledThreadPool(2); // 添加两个任务，希望它们都在 1s 后执行 executor.schedule(() -> { System.out.println("任务1，执行时间：" + new Date()); int i = 0 / 10; try { Thread.sleep(2000); } catch (InterruptedException e) { } }, 1000, TimeUnit.MILLISECONDS); executor.schedule(() -> { System.out.println("任务2，执行时间：" + new Date()); }, 1000, TimeUnit.MILLISECONDS);
可以看到，有异常的情况，输出并不受影响
任务2，执行时间：Mon Nov 06 16:59:26 CST 2023 任务1，执行时间：Mon Nov 06 16:59:26 CST 2023
timer中如果一个线程出现了异常，那么剩下的任务就不执行了，而ScheduledExecutorService出现了异常，那么后续任务还是会执行的
剑指JUC原理-13.线程池（下）:https://developer.aliyun.com/article/1413644
