1 线程池简介

  • 为什么要使用线程池（Thread Pool）？

    • 构造一个新的线程开销比较大，因为这涉及与操作系统的交互；
    • 当出现高并发时，如果为每个并发请求都创建一个线程，甚至可能导致内存溢出。
    • 因此，在大多数并发框架中都会使用线程池来管理线程

  • 使用线程池管理线程主要有如下好处

    • 降低资源消耗

通过复用已存在的线程和降低线程关闭的次数来尽可能降低系统性能损耗；

  • 提升系统响应速度

通过复用线程，省去创建线程的过程，因此整体上提升了系统的响应速度；

  • 提高线程的可管理性

线程是稀缺资源，如果无限制的创建，不仅会消耗系统资源，还会降低系统的稳定性，因此，需要使用线程池来管理线程。

  • 线程池的工作过程
    • 线程池中包含许多准备运行的线程
    • 为线程池提供一个Runnable，就会有一个线程调用其**run()**方法
    • 当**run()**方法退出时，这个线程不会死亡，而是留在池中准备为下一个请求提供服务

2 自定义线程池

  • 在了解Java提供的线程池时，首先了解一下自定义线程池，有助于后续的理解

  • 自定义线程池结构

image.png

  • 自定义线程池是一个生产者消费者模型，其中Thread Pool是消费者，而任务调用者是生产者
  • 需要一个阻塞队列来平衡平衡生产者与消费者速率的差异
  • 生产者不断将新的任务放到阻塞队列中，消费者则不断从阻塞队列中取出任务并处理
  • 自定义线程池代码 ```java //拒绝策略 @FunctionalInterface interface RejectPolicy { void reject(BlockingQueue queue, T task); }

@Slf4j(topic = “c.ThreadPool”) class ThreadPool { //任务队列 private final BlockingQueue taskQueue;

  1. //生产者：线程集合
  2. private final HashSet<Worker> workers = new HashSet<>();
  4. //核心线程数，即线程池中最大线程数量
  5. private final int coreSize;
  7. //任务的超时时间
  8. private final long timeout;
  9. private final TimeUnit timeUnit;
  11. private final RejectPolicy<Runnable> rejectPolicy;
  13. public ThreadPool(int coreSize, int queueCapacity, long timeout, TimeUnit timeUnit, RejectPolicy<Runnable> rejectPolicy) {
  14.     this.coreSize = coreSize;
  15.     this.timeout = timeout;
  16.     this.timeUnit = timeUnit;
  17.     this.taskQueue = new BlockingQueue<>(queueCapacity);
  18.     this.rejectPolicy = rejectPolicy;
  19. }
  21. //因为workers是共享变量，且不是线程安全的，因此加锁保护
  22. public void execute(Runnable task) {
  23.     synchronized (workers) {
  24.         //当线程池中的线程数量没有超过coreSize时，创建一个线程去执行任务
  25.         //否则任务加入阻塞队列暂存
  26.         if (workers.size() < coreSize) {
  27.             Worker worker = new Worker(task);
  28.             log.debug("新增 worker{}", worker);
  29.             workers.add(worker);
  30.             worker.start();
  31.         } else {
  32.             taskQueue.tryPut(rejectPolicy, task);
  33.         }
  34.     }
  35. }
  37. //内部类，表示线程池中的线程
  38. class Worker extends Thread {
  39.     private Runnable task;
  41.     public Worker(Runnable task) {
  42.         this.task = task;
  43.     }
  45.     @Override
  46.     public void run() {
  47.         //当task不为空，执行任务
  48.         //当task执行完毕，尝试从任务队列中获取任务继续执行
  49.         while (task != null || (task = taskQueue.take(timeout, timeUnit)) != null) {
  50.             try {
  51.                 log.debug("正在执行...{}", task);
  52.                 task.run();
  53.             } catch (Exception e) {
  54.                 e.printStackTrace();
  55.             } finally {
  56.                 task = null;
  57.             }
  58.         }
  60.         //已经没有任务了，移除当前线程
  61.         synchronized (workers) {
  62.             log.debug("worker 被移除{}", this);
  63.             workers.remove(this);
  64.         }
  65.     }
  66. }

}

@Slf4j(topic = “c.BlockingQueue”) class BlockingQueue { //任务队列 //双向队列有两种实现：LinkedList和ArrayDeque，通常ArrayDeque性能更好一些 private final Deque queue = new ArrayDeque<>();

  1. //锁
  2. private final ReentrantLock lock = new ReentrantLock();
  4. //生产者条件变量
  5. private final Condition fullWaitSet = lock.newCondition();
  7. //消费者条件变量
  8. private final Condition emptyWaitSet = lock.newCondition();
  10. //阻塞队列最大容量
  11. private final int capacity;
  13. public BlockingQueue(int capacity) {
  14.     this.capacity = capacity;
  15. }
  17. //根据timeout的值判断是超时获取还是获取
  18. public T take(long timeout, TimeUnit unit) {
  19.     //-1表示当没有任务时，线程会进入WAITING状态持续等待
  20.     if (timeout == -1)
  21.         //获取任务
  22.         return take();
  23.     //超时获取任务
  24.     return timed_take(timeout, unit);
  25. }
  27. //获取任务，如果没有任务则线程等待
  28. private T take() {
  29.     lock.lock();
  30.     try {
  31.         while (queue.isEmpty()) {
  32.             try {
  33.                 emptyWaitSet.await();
  34.             } catch (InterruptedException e) {
  35.                 e.printStackTrace();
  36.             }
  37.         }
  38.         T t = queue.removeFirst();
  39.         fullWaitSet.signal();
  40.         return t;
  41.     } finally {
  42.         lock.unlock();
  43.     }
  44. }
  46. //带超时的获取任务，如果没有任务则线程撤销
  47. private T timed_take(long timeout, TimeUnit unit) {
  48.     lock.lock();
  49.     try {
  50.         //将timeout统一转换为纳秒
  51.         long nanos = unit.toNanos(timeout);
  52.         while (queue.isEmpty()) {
  53.             try {
  54.                 //超时了，返回null并退出
  55.                 if (nanos <= 0) {
  56.                     return null;
  57.                 }
  58.                 //awaitNanos返回的是剩余等待时间
  59.                 nanos = emptyWaitSet.awaitNanos(nanos);
  60.             } catch (InterruptedException e) {
  61.                 e.printStackTrace();
  62.             }
  63.         }
  64.         T t = queue.removeFirst();
  65.         fullWaitSet.signal();
  66.         return t;
  67.     } finally {
  68.         lock.unlock();
  69.     }
  70. }
  72. //添加元素
  73. public void put(T element) {
  74.     lock.lock();
  75.     try {
  76.         while (queue.size() == capacity) {
  77.             try {
  78.                 log.debug("等待加入任务队列 {} ...", element);
  79.                 fullWaitSet.await();
  80.             } catch (InterruptedException e) {
  81.                 e.printStackTrace();
  82.             }
  83.         }
  84.         log.debug("加入任务队列 {}", element);
  85.         queue.addFirst(element);
  86.         emptyWaitSet.signal();
  87.     } finally {
  88.         lock.unlock();
  89.     }
  90. }
  92. public boolean timed_put(T element, long timeout, TimeUnit unit) {
  93.     lock.lock();
  94.     try {
  95.         long nanos = unit.toNanos(timeout);
  96.         while (queue.size() == capacity) {
  97.             try {
  98.                 log.debug("等待加入任务队列 {} ...", element);
  99.                 if (nanos <= 0)
  100.                     return false;
  101.                 nanos = fullWaitSet.awaitNanos(nanos);
  102.             } catch (InterruptedException e) {
  103.                 e.printStackTrace();
  104.             }
  105.         }
  106.         log.debug("加入任务队列 {}", element);
  107.         queue.addFirst(element);
  108.         emptyWaitSet.signal();
  109.     } finally {
  110.         lock.unlock();
  111.     }
  112.     return true;
  113. }
  115. public void tryPut(RejectPolicy<T> rejectPolicy, T element) {
  116.     lock.lock();
  117.     try {
  118.         //判断队列是否满
  119.         if (queue.size() == capacity) {
  120.             //当任务队列满时，将队列和任务交给rejectPolicy处理
  121.             rejectPolicy.reject(this, element);
  122.         } else {
  123.             log.debug("加入任务队列 {}", element);
  124.             queue.addFirst(element);
  125.             emptyWaitSet.signal();
  126.         }
  127.     } finally {
  128.         lock.unlock();
  129.     }
  130. }
  132. //获取阻塞队列当前大小
  133. public int size() {
  134.     lock.lock();
  135.     try {
  136.         return queue.size();
  137.     } finally {
  138.         lock.unlock();
  139.     }
  140. }

}

  2. - **测试**
  4. 任务队列容量为10，线程池容量为2，需要执行15个任务，拒绝策略为超时等待
  5. ```java
  6. @Slf4j(topic = "c.TestPool")
  7. public class TestPool {
  8.     public static void main(String[] args) {
  9.         //创建线程池并指定线程池容量、任务队列容量、任务队列空时线程的等待时间、任务队列满时的拒绝策略
  10.         ThreadPool threadPool = new ThreadPool(2, 10,
  11.                 3000, TimeUnit.MILLISECONDS, (queue, task) -> {
  12.             //1. 死等
  13.             //queue.put(task);
  14.             //2. 超时等待
  15.             if (!queue.timed_put(task, 1000, TimeUnit.MILLISECONDS))
  16.                 log.debug("等待超时，放弃{}", task);
  17.             //3. 调用者放弃任务执行
  18.             //log.debug("放弃{}", task);
  19.             //4. 调用者抛出异常
  20.             //throw new RuntimeException("任务执行失败" + task);
  21.             //5. 调用者自己执行任务
  22.             //task.run();
  23.         });
  25.         //创建15个任务
  26.         for (int i = 0; i < 15; i++) {
  27.             int temp = i;
  28.             threadPool.execute(() -> {
  29.                 try {
  30.                     Thread.sleep(5000);
  31.                 } catch (InterruptedException e) {
  32.                     e.printStackTrace();
  33.                 }
  34.                 log.debug("{}", temp);
  35.             });
  36.         }
  37.     }
  38. }
  • 上述代码输出
    1. 16:28:11 [main] c.ThreadPool - 新增 workerThread[Thread-0,5,main]
    2. 16:28:11 [main] c.ThreadPool - 新增 workerThread[Thread-1,5,main]
    3. 16:28:11 [Thread-0] c.ThreadPool - 正在执行...TestPool$$Lambda$31/0x00000008000cb840@4ae63b4b
    4. 16:28:11 [Thread-1] c.ThreadPool - 正在执行...TestPool$$Lambda$31/0x00000008000cb840@577d4f43
    5. 16:28:11 [main] c.BlockingQueue - 加入任务队列 TestPool$$Lambda$31/0x00000008000cb840@79efed2d
    6. 16:28:11 [main] c.BlockingQueue - 加入任务队列 TestPool$$Lambda$31/0x00000008000cb840@27ae2fd0
    7. 16:28:11 [main] c.BlockingQueue - 加入任务队列 TestPool$$Lambda$31/0x00000008000cb840@29176cc1
    8. 16:28:11 [main] c.BlockingQueue - 加入任务队列 TestPool$$Lambda$31/0x00000008000cb840@2f177a4b
    9. 16:28:11 [main] c.BlockingQueue - 加入任务队列 TestPool$$Lambda$31/0x00000008000cb840@4278a03f
    10. 16:28:11 [main] c.BlockingQueue - 加入任务队列 TestPool$$Lambda$31/0x00000008000cb840@147ed70f
    11. 16:28:11 [main] c.BlockingQueue - 加入任务队列 TestPool$$Lambda$31/0x00000008000cb840@61dd025
    12. 16:28:11 [main] c.BlockingQueue - 加入任务队列 TestPool$$Lambda$31/0x00000008000cb840@124c278f
    13. 16:28:11 [main] c.BlockingQueue - 加入任务队列 TestPool$$Lambda$31/0x00000008000cb840@15b204a1
    14. 16:28:11 [main] c.BlockingQueue - 加入任务队列 TestPool$$Lambda$31/0x00000008000cb840@77167fb7
    15. 16:28:11 [main] c.BlockingQueue - 等待加入任务队列 TestPool$$Lambda$31/0x00000008000cb840@1fe20588 ...
    16. 16:28:12 [main] c.BlockingQueue - 等待加入任务队列 TestPool$$Lambda$31/0x00000008000cb840@1fe20588 ...
    17. 16:28:12 [main] c.TestPool - 等待超时，放弃TestPool$$Lambda$31/0x00000008000cb840@1fe20588
    18. 16:28:12 [main] c.BlockingQueue - 等待加入任务队列 TestPool$$Lambda$31/0x00000008000cb840@6973bf95 ...
    19. 16:28:13 [main] c.BlockingQueue - 等待加入任务队列 TestPool$$Lambda$31/0x00000008000cb840@6973bf95 ...
    20. 16:28:13 [main] c.TestPool - 等待超时，放弃TestPool$$Lambda$31/0x00000008000cb840@6973bf95
    21. 16:28:13 [main] c.BlockingQueue - 等待加入任务队列 TestPool$$Lambda$31/0x00000008000cb840@2ddc8ecb ...
    22. 16:28:14 [main] c.BlockingQueue - 等待加入任务队列 TestPool$$Lambda$31/0x00000008000cb840@2ddc8ecb ...
    23. 16:28:14 [main] c.TestPool - 等待超时，放弃TestPool$$Lambda$31/0x00000008000cb840@2ddc8ecb
    24. 16:28:16 [Thread-1] c.TestPool - 1
    25. 16:28:16 [Thread-0] c.TestPool - 0
    26. 16:28:16 [Thread-1] c.ThreadPool - 正在执行...TestPool$$Lambda$31/0x00000008000cb840@77167fb7
    27. 16:28:16 [Thread-0] c.ThreadPool - 正在执行...TestPool$$Lambda$31/0x00000008000cb840@15b204a1
    28. 16:28:21 [Thread-1] c.TestPool - 11
    29. 16:28:21 [Thread-0] c.TestPool - 10
    30. 16:28:21 [Thread-0] c.ThreadPool - 正在执行...TestPool$$Lambda$31/0x00000008000cb840@61dd025
    31. 16:28:21 [Thread-1] c.ThreadPool - 正在执行...TestPool$$Lambda$31/0x00000008000cb840@124c278f
    32. 16:28:26 [Thread-0] c.TestPool - 8
    33. 16:28:26 [Thread-1] c.TestPool - 9
    34. 16:28:26 [Thread-0] c.ThreadPool - 正在执行...TestPool$$Lambda$31/0x00000008000cb840@147ed70f
    35. 16:28:26 [Thread-1] c.ThreadPool - 正在执行...TestPool$$Lambda$31/0x00000008000cb840@4278a03f
    36. 16:28:31 [Thread-1] c.TestPool - 6
    37. 16:28:31 [Thread-0] c.TestPool - 7
    38. 16:28:31 [Thread-1] c.ThreadPool - 正在执行...TestPool$$Lambda$31/0x00000008000cb840@2f177a4b
    39. 16:28:31 [Thread-0] c.ThreadPool - 正在执行...TestPool$$Lambda$31/0x00000008000cb840@29176cc1
    40. 16:28:36 [Thread-1] c.TestPool - 5
    41. 16:28:36 [Thread-0] c.TestPool - 4
    42. 16:28:36 [Thread-1] c.ThreadPool - 正在执行...TestPool$$Lambda$31/0x00000008000cb840@27ae2fd0
    43. 16:28:36 [Thread-0] c.ThreadPool - 正在执行...TestPool$$Lambda$31/0x00000008000cb840@79efed2d
    44. 16:28:42 [Thread-1] c.TestPool - 3
    45. 16:28:42 [Thread-0] c.TestPool - 2
    46. 16:28:45 [Thread-0] c.ThreadPool - worker 被移除Thread[Thread-0,5,main]
    47. 16:28:45 [Thread-1] c.ThreadPool - worker 被移除Thread[Thread-1,5,main]

3 ThreadPoolExecutor类

  • ThreadPoolExecutor概述

ThreadPoolExecutor类是JDK提供的线程池实现，位于JUC包

1 线程池状态

  • 状态表示
    • ThreadPoolExecutor使用int的高3位来表示线程池状态，低29位表示线程数量
    • 这样表示的目的是将线程池状态与线程个数合二为一，这样就可以用一次CAS原子操作来改变线程池状态
    • 上述信息存储在一个原子变量ctl中 ```java //ctl存储线程状态和线程个数 private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

//rs为int高3位，代表线程池状态；wc为int低29位，代表线程格式；该方法用于合并rs和wc private static int ctlOf(int rs, int wc) { return rs | wc; }

  2. - **线程池状态表**
  3. | **状态名** | **高3位** | **接收新任务** | **处理阻塞队列任务** | **说明** |
  4. | --- | --- | --- | --- | --- |
  5. | **RUNNING** | 111 | Y | Y | 线程池新创建时的状态，可以接受新任务，也可以处理阻塞队列任务 |
  6. | **SHUTDOWN** | 000 | N | Y | 不会接收新任务，但会处理阻塞队列剩余任务 |
  7. | **STOP** | 001 | N | N | 会中断正在执行的任务，并抛弃阻塞队列中的任务 |
  8. | **TIDYING** | 010 | <br /> | <br /> | 任务全部执行完毕，没有活动线程，即将进入终结 |
  9. | **TERMINATED** | 011 | <br /> | <br /> | 终结 |
  11. <br />
  13. <a name="QmOra"></a>
  14. ## 2 线程池创建（参数与行为）
  16. - 从自定义线程池一节可以了解到，线程池的创建需要配置很多参数，而通过ThreadPoolExecutor类创建线程池也是一样的，其有许多重载的构造方法
  18. - 以参数最多的构造方法来理解创建线程池的**7个参数**
  19. ```java
  20. public ThreadPoolExecutor(int corePoolSize,
  21.                           int maximumPoolSize,
  22.                           long keepAliveTime,
  23.                           TimeUnit unit,
  24.                           BlockingQueue<Runnable> workQueue,
  25.                           ThreadFactory threadFactory,
  26.                           RejectedExecutionHandler handler)

可以发现ThreadPoolExecutor的参数配置十分类似于自定义线程池的参数配置，具体的参数配置含义如下
**int corePoolSize**

  • 表示核心线程池的大小
  • 线程池中的线程分为两类，核心线程和救急线程，这里指定的是核心线程的数量
  • 当提交一个任务时，如果当前核心线程的个数没有达到corePoolSize，即使当前核心线程池有空闲的线程，也会创建新的核心线程来执行所提交的任务。如果当前核心线程池的线程个数已经达到了corePoolSize，则不再重新创建线程。
  • 当线程数达到corePoolSize并且没有线程空闲，这时新加入的任务会被加入阻塞队列，直到有空闲的线程
  • 当核心线程执行完任务后，也不会被销毁
  • 核心线程默认“懒启动”，即线程池创建时没有线程

如果调用了prestartCoreThread()或者prestartAllCoreThreads()，线程池创建的时候所有的核心线程都会被创建并且启动

**int maximumPoolSize**

  • 表示线程池能创建线程的最大个数，超过corePoolSize的线程称为救急线程
  • 当选择了有界阻塞队列且阻塞队列已满，并且当前线程池线程个数没有超过maximumPoolSize，就会创建救急线程来执行任务
  • 救急线程不像核心线程一样创建后就在线程池中长期存在，而是会在无事可做后等待一定时间，最后被销毁

**long keepAliveTime**

  • 表示救急线程存活的时间，这里的存活时间指的是当线程无事可做时等待新任务的时间
  • 如果当前线程池的线程个数已经超过了corePoolSize，并且线程空闲时间超过了keepAliveTime的话，就会将这些救急线程销毁，这样可以尽可能降低系统资源消耗。

**TimeUnit unit**

  • 表示时间单位，为keepAliveTime指定时间单位。

**BlockingQueue<Runnable> workQueue**

  • 表示阻塞队列
  • BlockingQueue是一个接口，JDK提供了7种实现类，常用的4种如下
    • ArrayBlockingQueue

ArrayBlockingQueue是用数组实现的有界阻塞队列，其容量在创建后不可改变
队列中的元素执行先进先出策略

  1.   - **LinkedBlockingQueue**

LinkedBlockingQueue是用链表实现的有界阻塞队列，其内元素执行先进先出策略

  1.   - **SynchronousQueue**

SynchronousQueue每个插入任务操作前必须等待线程池中的线程准备来阻塞队列中取任务
因此，SynchronousQueue实际上没有存储任何数据元素，换句话来说生产者与消费者直接交易

  1.   - **PriorityBlockingQueue**

支持优先级排序的阻塞队列

**ThreadFactory threadFactory**

  • 表示创建线程的线程工厂
  • ThreadFactory是一个接口，其内只有一个方法Thread newThread(Runnable r)

实现ThreadFactory接口主要是为每个创建出来的线程设置更有意义的名字

  • 默认调用Executors类中的静态工厂方法_**defaultThreadFactory**_**()**来创建一个默认的线程工厂

    1. public ThreadPoolExecutor(int corePoolSize,
    2.                      int maximumPoolSize,
    3.                      long keepAliveTime,
    4.                      TimeUnit unit,
    5.                      BlockingQueue<Runnable> workQueue) {
    6. this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
    7.     Executors.defaultThreadFactory(), defaultHandler);
    8. }
    1. //Executors类中的静态工厂方法
    2. public static ThreadFactory defaultThreadFactory() {
    3. return new DefaultThreadFactory();
    4. }
    1. //默认线程工厂
    2. private static class DefaultThreadFactory implements ThreadFactory {
    3. private static final AtomicInteger poolNumber = new AtomicInteger(1);
    4. private final ThreadGroup group;
    5. private final AtomicInteger threadNumber = new AtomicInteger(1);
    6. private final String namePrefix;
    8. DefaultThreadFactory() {
    9.    SecurityManager s = System.getSecurityManager();
    10.    group = (s != null) ? s.getThreadGroup() :
    11.                          Thread.currentThread().getThreadGroup();
    12.    namePrefix = "pool-" +
    13.                  poolNumber.getAndIncrement() +
    14.                 "-thread-";
    15. }
    17. public Thread newThread(Runnable r) {
    18.    Thread t = new Thread(group, r,
    19.                          namePrefix + threadNumber.getAndIncrement(),
    20.                          0);
    21.    if (t.isDaemon())
    22.        t.setDaemon(false);
    23.    if (t.getPriority() != Thread.NORM_PRIORITY)
    24.        t.setPriority(Thread.NORM_PRIORITY);
    25.    return t;
    26. }
    27. }

**RejectedExecutionHandler handler**

  • 表示拒绝策略
  • 当线程池的阻塞队列已满并且所有线程都已开启（当前线程数量达到maximumPoolSize），说明当前线程池已经处于饱和状态了，那么就需要采用一种拒绝策略来处理所提交的任务
  • RejectedExecutionHandler是一个接口，JDK提供了4种实现类（拒绝策略），如下
    • AbortPolicy

直接拒绝所提交的任务，并抛出RejectedExecutionException异常，这是默认拒绝策略

  1.   - **CallerRunsPolicy**

利用调用者所在的线程来执行任务；

  1.   - **DiscardPolicy**

不处理直接丢弃掉任务

  1.   - **DiscardOldestPolicy**

丢弃掉阻塞队列中存放时间最久的任务，执行当前任务
这些实现类都是ThreadPoolExecutor下的内部静态类，也可以根据自己的需要实现自己的拒绝策略，只需要实现RejectedExecutionHandler接口即可

Executors类

  • Executors类概述
    • ThreadPoolExecutor类的构造方法参数众多，比较难掌握，因此JDK在Executors类中提供了众多工厂方法来方便的创建线程池
    • Executors类的工程方法内部实际上调用的仍然是ThreadPoolExecutor类的构造方法，只不过简化了参数的配置
  1. **ExecutorService newFixedThreadPool(int nThreads)**

  • 源码

    1. public static ExecutorService newFixedThreadPool(int nThreads) {
    2.   return new ThreadPoolExecutor(nThreads, nThreads,
    3.                                 0L, TimeUnit.MILLISECONDS,
    4.                                 new LinkedBlockingQueue<Runnable>());
    5. }

  • 特点

    • 核心线程数 == 最大线程数（没有救急线程被创建），因此也无需超时时间
    • 适用于任务量已知，相对耗时的任务
  1. **ExecutorService newCachedThreadPool()**

  • 源码

    1. public static ExecutorService newCachedThreadPool() {
    2.   return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
    3.                                 60L, TimeUnit.SECONDS,
    4.                                 new SynchronousQueue<Runnable>());
    5. }

  • 特点

    • 核心线程数是0， 最大线程数是Integer.MAX_VALUE，救急线程的空闲生存时间是60s，意味着线程池中的线程全部都是救急线程且60s后可以回收
    • 整个线程池表现为线程数会根据任务量不断增长，没有上限；当任务执行完毕，空闲1分钟后释放线程。
    • 适合任务数比较密集，但每个任务执行时间较短的情况
  1. **ExecutorService newSingleThreadExecutor()**

  • 源码

    1. public static ExecutorService newSingleThreadExecutor() {
    2.   return new FinalizableDelegatedExecutorService
    3.       (new ThreadPoolExecutor(1, 1,
    4.                               0L, TimeUnit.MILLISECONDS,
    5.                               new LinkedBlockingQueue<Runnable>()));
    6. }

  • 特点

希望多个任务排队执行。线程数固定为1，任务数多于1时，会放入阻塞队列排队。任务执行完毕，这唯一的线程也不会被释放

  • 自己创建一个线程与该线程池的区别
    • 自己创建一个单线程串行执行任务，如果任务执行失败那么线程将会结束而剩余任务无法完成
    • 使用该线程池时如果其内线程意外终止，线程池还会新建一个线程，保证池的正常工作

3 Callable接口与Future接口

  • Callable和Future是位于JUC包下的接口

  • Callable接口概述

    • Runnable接口用于封装线程所执行的任务，其内提供一个没有参数和返回值的方法run()
    • Callable接口也是用于封装线程所执行的任务，其内提供一个没有参数但是有返回值的方法call()
    • Callable接口源码如下

Callable接口的泛型参数是返回值的类型，如**Callable<Integer>**表示一个返回Integer对象的任务

  1. @FunctionalInterface
  2. public interface Callable<V> {
  3.     V call() throws Exception;
  4. }
  • Future接口概述
    • Future用于保存异步计算的结果
    • 一般来说，当我们执行一个长时间运行的任务时，使用Future就可以让我们暂时去处理其他的任务，等长任务执行完毕再返回其结果
    • Future接口源码如下

Future接口的泛型参数是get()方法返回值的类型

  1. public interface Future<V> {
  2.     boolean cancel(boolean mayInterruptIfRunning);
  3.     boolean isCancelled();
  4.     boolean isDone();
  5.     V get() throws InterruptedException, ExecutionException;
  6.     V get(long timeout, TimeUnit unit)
  7.         throws InterruptedException, ExecutionException, TimeoutException;
  8. }

  • 执行Callable的方式

    1. 可以使用FutureTask类来包裹一个Callable，由于FutureTask类实现了Future接口和Runnable接口，因此可以将FutureTask传给一个线程来执行Callable的**call()**方法

    • FutureTask相关源码

      1. public class FutureTask<V> implements Runnable, Future<V> {
      3. private Callable<V> callable;
      5. public FutureTask(Callable<V> callable) {
      6.    if (callable == null)
      7.        throw new NullPointerException();
      8.    this.callable = callable;
      9.    this.state = NEW;       // ensure visibility of callable
      10. }
      12. // 简略版
      13. public void run() {
      14.    ...
      15.    Callable<V> c = callable;
      16.    V result;
      17.    result = c.call();    // 执行Callable的call方法
      18.    ...
      19. }
      20. }
    1. 更常见的是直接将Callable传给线程池的方法，由线程池来执行任务

Future接口API

  • Future的方法都是实例方法

  • **V future.get() throws InterruptedException, ExecutionException**

  • **V future.get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException**

该方法会阻塞当前线程，使当前线程进入WAITING状态或TIMED_WAITING状态，直到结果可用或超过了指定的时间，类似于保护性暂停模式

  • 如果超过了指定的时间，将抛出TimeoutException异常
  • **boolean future.cancel(boolean mayInterruptIfRunning)**

尝试取消任务的运行

  • 当任务还没有开始，则取消任务且不再开始
  • 如果任务已经开始且参数mayInterruptIfRunning为true，则中断任务


  • **boolean future.isDone()**

如果任务结束（无论是正常完成、中途取消还是发生异常），都返回true

4 线程池提交任务相关API

  • 所有API都是实例方法

  • **void pool.execute(Runnable command)**

    • 向线程池提交任务并由线程池执行

  • **Future<?> pool.submit(Runnable task)**

  • **<T> Future<T> pool.submit(Callable<T> task)**

与execute()方法功能类似，不同的是

  • **submit()**方法既可以接收**Runnable**类型也可以接收**Callable<T>**类型，
  • **execute()**方法没有返回值，而**submit()**方法会返回一个Future对象，通过它可以判断任务是否执行成功
  • 获取任务执行结果时可以调用Future::get()，该方法会阻塞当前线程直到任务完成

  • 实例 ```java @Slf4j() public class Test { public static void main(String[] args) throws Exception { ExecutorService pool = Executors.newFixedThreadPool(3);

    Future future = pool.submit(new Callable() {

    1.    @Override
    2.    public String call() throws Exception {
    3.        log.debug("running");
    4.        Thread.sleep(1000);
    5.        return "ok";
    6.    }

    });

    log.debug(“{}”, future.get()); } }

  1. 由于Future类的`get()`方法内部有保护性暂停策略，因此上述主线程打印时一定在子线程执行完毕后
  3. - `**<T> List<Future<T>> pool.invokeAll(Collection<? extends Callable<T>> tasks)**`
  4. - `**<T> List<Future<T>> pool.invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)**`
  5.    - 向线程池提交**tasks中所有任务**
  6.    - 可以选择向该方法传递超时参数，当执行超时时，线程池将取消执行tasks中剩余的任务
  8. - `**<T> T pool.invokeAny(Collection<? extends Callable<T>> tasks)**`
  9.    - 提交tasks中所有任务，**哪个任务先成功执行完毕，返回此任务执行结果，其它任务取消**
  11. <a name="mO3fT"></a>
  12. ## 5 关闭线程池相关API
  14. - 所有API都是**实例方法**
  16. - `**void pool.shutdown()**`
  17.    - 将线程池状态切换为**SHUTDOWN**
  18.    - 此时不会接收新任务，但**已提交的任务会执行完**
  20. - `**List<Runnable> pool.shutdownNow()**`
  21.    - 将线程池状态切换为**STOP**
  22.    - 此时不接收新任务，**中断正在执行的任务，返回阻塞队列中任务**
  24. <a name="pq7bh"></a>
  25. # 4 Fork-Join池
  27. - **Fork-Join池概述**
  28.    - Fork-Join是JDK7加入的**新的线程池实现**
  29.    - Fork-Join体现的是一种**分治思想**，**适用于能够进行任务拆分的CPU密集型运算，I/O密集型任务仍然需要使用传统的线程池**
  30.    - 任务拆分是指将一个大任务拆分为算法上相同的小任务，直至不能拆分可以直接求解。
  32. 跟**递归相关的一些计算**，如归并排序、斐波那契数列、都可以用分治思想进行求解<br />换句话说，**Fork-Join池可以自动地以并行的方式处理递归任务**
  34.    - **Fork-Join在分治的基础上加入了多线程**，可以**把每个任务的分解和合并交给不同的线程来完成**，进一步提升了运算效率
  36. <br />
  38. - **Fork-Join池的创建**
  39.    - Fork-Join池在JDK中的实现是**ForkJoinPool类**，其位于JUC包下
  40.    - ForkJoinPool的创建类似于ThreadPoolExecutor，同样有很多参数和重载的构造方法
  41.    - Fork-Join池默认会创建与**CPU核心数大小相同的线程池
  43. **
  44. - **Fork-Join池的任务的创建**
  45.    - 提交给Fork-Join池的任务，需要**继承**`**RecursiveTask<T>**`**类（任务有返回值）**或`**RecursiveAction**`**类（任务无返回值）**
  46.    - 具体执行的任务逻辑在上述两个类的`compute()`方法中，因此需要重写该方法
  47.    - `compute()`中的任务逻辑类似于递归任务的逻辑，稍有不同
  49. - **Fork-Join池的使用**
  51. **实例**
  53.    - 统计一个数组中有多少个元素满足大于0
  54.    - 我们可以使用二分的思想，将数组一分为二，分别对这两部分进行统计，再将结果相加
  55. ```java
  56. public class Test {
  57.     public static void main(String[] args) throws Exception {
  58.         int[] a = {-1, 1, -1, 1, -1, 1, -1, 1, -1, 1};
  59.         Counter counter = new Counter(a, 0, a.length);
  60.         ForkJoinPool pool = new ForkJoinPool();
  61.         pool.invoke(counter);
  62.         System.out.println(counter.join());
  63.     }
  64. }
  66. class Counter extends RecursiveTask<Integer> {
  67.     public static final int THRESHOLD = 3;  //任务拆分阈值
  68.     private final int[] values;
  69.     private final int from;
  70.     private final int to;
  72.     public Counter(int[] values, int from, int to) {
  73.         this.values = values;
  74.         this.from = from;
  75.         this.to = to;
  76.     }
  78.     @Override
  79.     protected Integer compute() {
  80.         if (to - from < THRESHOLD) {
  81.             //不再拆分，直接解决问题
  82.             int count = 0;
  83.             for (int i = from; i < to; i++) {
  84.                 if (values[i] > 0) count++;
  85.             }
  86.             return count;
  87.         } else {
  88.             int mid = (from + to) / 2;
  89.             Counter first = new Counter(values, from, mid);
  90.             Counter second = new Counter(values, mid, to);
  91.             first.fork();   // 让一个线程执行该任务
  92.             second.fork();
  93.             return first.join() + second.join();    //合并结果
  94.         }
  95.     }
  96. }