1. 使用场景

  • 单个任务处理时间较短，需要处理的任务数量很大；

    2. 使用线程池的好处

  • 降低资源消耗，通过重复利用已创建的线程降低线程创建和销毁造成的消耗；

  • 提高响应速度，当任务到达时，任务可以不需要的等到线程创建就能立即执行；

  • 提高线程的可管理性，线程是稀缺资源，如果无限制的创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一的分配，调优和监控；

    3. jdk 自带线程池类图

    Java 线程池 - 图1

  • Executor 接口

    • 对于不同的 Executor 实现，execute() 方法可能是创建一个新线程并立即启动，也有可能是使用已有的工作线程来运行传入的任务，也可能是根据设置线程池的容量或者阻塞队列的容量来决定是否要将传入的线程放入阻塞队列中或者拒绝接收传入的线程；
  • ExecutorService 接口
    • ExecutorService 接口继承自 Executor 接口，提供了管理终止的方法，以及可为跟踪一个或多个异步任务执行状况而生成 Future 的方法。增加了 shutDown()，shutDownNow()，invokeAll()，invokeAny()和submit() 等方法。如果需要支持即时关闭，也就是 shutDownNow() 方法，则任务需要正确处理中断；

  • ScheduledExecutorService 接口

    • ScheduledExecutorService 扩展 ExecutorService 接口并增加了 schedule 方法。调用 schedule 方法可以在指定的延时后执行一个 Runnable 或者 Callable 任务。ScheduledExecutorService 接口还定义了按照指定时间间隔定期执行任务的 scheduleAtFixedRate() 方法和 scheduleWithFixedDelay() 方法；

      4. 线程池的创建

      4.1. ThreadPoolExecutor

  • 构造函数

    1. public ThreadPoolExecutor(  
    2.   int corePoolSize,  // 队列没满时，线程最大并发数
    3.   int maximumPoolSize,  // 队列满后线程能够达到的最大并发数
    4.   long keepAliveTime, // 空闲线程过多久被回收的时间限制
    5.   TimeUnit unit, // keepAliveTime 的时间单位
    6.   BlockingQueue<Runnable> workQueue, // 阻塞的队列类型
    7.   ThreadFactory threadFactory,
    8.   RejectedExecutionHandler handler // 超出 maximumPoolSizes + workQueue 时，任务会交给RejectedExecutionHandler来处理
    9. )

  • 向线程池提交任务

Java 线程池 - 图2

  • 补充： RunnableTaskQueue 用于保存等待执行的任务的阻塞队列

    • ArrayBlockingQueue：是一个基于数组结构的有界阻塞队列，此队列按 FIFO原则对元素进行排序；
    • LinkedBlockingQueue：一个基于链表结构的无界阻塞队列，此队列按FIFO排序元素，吞吐量通常要高于ArrayBlockingQueue，应用： nexFixedThreadPool；
    • SynchronousQueue：一个不存储元素的阻塞队列，每个插入操作必须等到另一个线程调用移除操作，否则插入操作一直处于阻塞状态，吞吐量通常要高于 LinkedBlockingQueue；
    • PriorityBlockingQueue：一个具有优先级得无限阻塞队列；

      4.2. newFixedThreadPool

  • 实现：创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待；

  • 构造函数

    1. public static ExecutorService newFixedThreadPool(int nThreads){
    2.   return new ThreadPoolExecutor(
    3.           nThreads,   // corePoolSize
    4.           nThreads,   // maximumPoolSize == corePoolSize
    5.           0L,         // 空闲时间限制是 0
    6.           TimeUnit.MILLISECONDS,
    7.           new LinkedBlockingQueue<Runnable>() // 无界阻塞队列
    8.       );
    9. }
    10. ExecutorService fixPool = Executors.newFixedThreadPool(corePoolSize);

  • 向线程池提交任务 Java 线程池 - 图3

    4.3. newCachedThreadPool

  • 实现： 创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收则新建线程；

  • 构造函数

    1. public static ExecutorService newCachedThreadPool(){
    2.   return new ThreadPoolExecutor(
    3.       0,                  // corePoolSoze == 0
    4.       Integer.MAX_VALUE,  // maximumPoolSize 非常大
    5.       60L,                // 空闲判定是60 秒
    6.       TimeUnit.SECONDS,
    7.       // 神奇的无存储空间阻塞队列，每个 put 必须要等待一个 take
    8.       new SynchronousQueue<Runnable>()  
    9.   );
    10. }
    11. ExecutorService cachePool = Executors.newCachedThreadPool();

  • 向线程池提交任务

Java 线程池 - 图4

4.4. newSingleThreadExecutor

  • 实现：创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行；

  • 构造函数

    1. public static ExecutorService newSingleThreadExecutor() {
    2.       return 
    3.           new FinalizableDelegatedExecutorService
    4.               (
    5.                   new ThreadPoolExecutor
    6.                       (
    7.                           1,
    8.                           1,
    9.                           0L,
    10.                           TimeUnit.MILLISECONDS,
    11.                           new LinkedBlockingQueue<Runnable>(),
    12.                           threadFactory
    13.                       )
    14.               );
    15. }
    16. ExecutorService singPool = Executors.newSingleThreadExecutor();

    除了多了个 FinalizableDelegatedExecutorService 代理，其初始化和 newFiexdThreadPool 的 nThreads = 1 的时候是一样的。 区别在于：

    • newSingleThreadExecutor 返回的 ExcutorService 在析构函数finalize()处会调用shutdown();
    • 如果我们没有对它调用 shutdown()，那么可以确保它在被回收时调用 shutdown() 来终止线程;
    • 使用 ThreadFactory，可以改变线程的名称、线程组、优先级、守护进程状态，一般采用默认;

      4.5 newScheduledThreadPool

  • 定时线程池，该线程池可用于周期性地去执行任务，通常用于周期性的同步数据；

    1. public static ScheduledExecutorService newScheduledThreadPool(int var0) {
    2.       return new ScheduledThreadPoolExecutor(var0);
    3.   }
    4.   public static ScheduledExecutorService newScheduledThreadPool(int var0, ThreadFactory var1) {
    5.       return new ScheduledThreadPoolExecutor(var0, var1);
    6.   }

    。使用

    • scheduleAtFixedRate：以固定频率执行任务，周期是指每次执行任务成功执行之间的间隔；
    • schedultWithFixedDelay：以固定延时执行任务，延时是指上一次执行成功后和下一次开始执行前的时间；
      1. public class ScheduledExecutorServiceDemo {
      2. public static void main(String args[]) {
      3.    ScheduledExecutorService ses = Executors.newScheduledThreadPool(10);
      4.    ses.scheduleAtFixedRate(new Runnable() {
      5.        @Override
      6.        public void run() {
      7.            try {
      8.                Thread.sleep(4000);
      9.                System.out.println(Thread.currentThread().getId() + "执行了");
      10.            } catch (InterruptedException e) {
      11.                e.printStackTrace();
      12.            }
      13.        }
      14.    }, 0, 2, TimeUnit.SECONDS);
      15. }
      16. }

      5. 其他

      5.1 关闭线程池

  • shutdown : 不会立即终止线程池，而是将线程池的状态设置成 SHUTDOWN 状态，然后中断所有没有正在执行任务的线程，继续将缓存队列中的人物执行完后才终止；;

  • shutdownNow :
    • 将线程池的状态设置成 STOP ;
    • 遍历线程池中的工作线程，逐个调用线程的 interrupt 方法来中断线程;
    • 无法响应中断的任务可能永远无法终止；
    • 总结：立即终止线程池，并尝试打断正在执行的任务，清空任务缓存队列，返回尚未执行的任务；

  • 所有任务均已关闭，说明线程池关闭成功，调用 isTerminaed 方法会返回 true；

    5.2 监控线程池

  • taskCount：线程池需要执行的任务数量；

  • completedTaskCount：线程池在运行过程中已完成的任务数量，不大于 taskCount；
  • largestPoolSize：线程池曾创建的最大线程数量，通过此参数可判断线程池是否满过；
  • getPoolSize：线程池的线程数量；若线程池不销毁，池里线程不会自动销毁；

  • getActiveCount：获取活动的线程数；

    5.3 提交任务 execute 与 submit 的区别

  • execute 方法无返回值，无法判断任务是否被线程池执行成功；

  • submit 方法会返回一个 future 对象，future.get() 方法获取返回值；

    6. 如何选择

    6.1 如何选择线程池数量

  • 线程池的大小决定着系统的性能，过大或者过小的线程池数量都无法发挥最优的系统性能；

  • 考虑因素： CPU 的数量，内存大小，任务类型（eg. 计算密集型、IO密集型等）
  • 一般公式 ```java NCPU = CPU的数量 UCPU = 期望对CPU的使用率 0 ≤ UCPU ≤ 1 W/C = 等待时间与计算时间的比率

如果希望处理器达到理想的使用率，那么线程池的最优大小为： 线程池大小= NCPU *UCPU(1+W/C)

int ncpus = Runtime.getRuntime().availableProcessors();//获取CPU的数量

  1. <a name="aS4el"></a>
  2. ### 6.2 线程池创建方式
  4. - 线程池不允许使用 Executors 去创建，而是通过 ThreadPoolExecutor 的方式【阿里编码规范】；
  5. - Executors 各个方法的弊端：
  6.    - newFixedThreadPool 和 newSingleThreadExecutor
  7.       - 主要问题是堆积的请求处理队列可能会耗费非常大的内存，甚至OOM；
  8.    - newCachedThreadPool和newScheduledThreadPool
  9.       - 主要问题是线程数最大数是 Integer.MAX_VALUE，可能会创建数量非常多的线程，甚至OOM；
  10. <a name="KNfgJ"></a>
  11. ## 7. 手写一个简单线程池
  12. ```java
  13. import java.util.concurrent.ArrayBlockingQueue;
  14. import java.util.concurrent.BlockingQueue;
  15. import java.util.concurrent.atomic.AtomicInteger;
  17. /** 参考链接： https://yq.aliyun.com/articles/680584
  18.  * 简单的手写线程池，貌似是针对 coreSize = maxSize 做的处理，而对 maxSize > coreSize 的情况没有处理，
  19.  * 目前运行无终止，后续继续优化；
  20.  */
  21. public class testThreadPool {
  23.     public static void main(String[] args ){
  24.         CustomThreadPool();
  25.     }
  27.     public static void CustomThreadPool(){
  28.         TestPool testPool = new TestPool(2,2,new ArrayBlockingQueue<>(3));
  29.         for (int i = 0; i < 100; i++) {
  30.             final int j = i;
  31.             System.out.println("i = " + i + " " + Thread.currentThread().getName());
  32.             testPool.execute(()->{
  33.                 try {
  34.                     Thread.sleep(100);
  35.                     System.out.println("睡 0.1 秒，完成 ："  + j);
  36.                 } catch (InterruptedException e) {
  37.                     e.printStackTrace();
  38.                 }
  39.             });
  40.         }
  41.     }
  42. }
  44. class TestPool{
  46.     private int coreSize;
  47.     private int maxSize;
  48.     private AtomicInteger running = new AtomicInteger(0);
  49.     private BlockingQueue<Runnable> queue;
  51.     public TestPool(int coreSize, int maxSize, ArrayBlockingQueue<Runnable> queue){
  52.         this.coreSize = coreSize;
  53.         this.maxSize = maxSize;
  54.         this.queue = queue;
  55.     }
  57.     public void execute(Runnable runnable){
  58.         if (running.get() < coreSize) {
  59.             if (!addWorker(runnable)) {
  60.                 reject();
  61.             }
  62.         } else {
  63.             System.out.println("当前队列大小： " + queue.size());
  64.             if (!queue.offer(runnable)) {
  65.                 System.out.println("offer失败，当前线程数： " + running.get());//offer是有返回boolean类型的，put()不可以；
  66.                 if (!addWorker(runnable)) {
  67.                     reject();
  68.                 }
  69.             }
  70.         }
  71.     }
  73.     private void reject(){
  74.         throw new RuntimeException("超出大小，线程数： " + running.get() + " , 队列大小 ： " + queue.size());
  75.     }
  77.     private boolean addWorker(Runnable runnable){
  78.         if (running.get() >= maxSize) {
  79.             return false;
  80.         }
  81.         Worker worker = new Worker(runnable);
  82.         worker.start();
  83.         return true;
  84.     }
  86.     private class Worker extends Thread{
  87.         private Runnable runnable;
  88.         public Worker(Runnable runnable){
  89.             this.runnable = runnable;
  90.             System.out.println("创建线程，当前线程数 ： " + running.incrementAndGet());
  91.         }
  92.         @Override
  93.         public void run() {
  94.             try{
  95.                 while (true) {
  96.                     runnable.run();
  97.                     System.out.println("运行结束，当前线程数 ： " + running.get());
  98.                     if (running.get() > coreSize) {
  99.                         break;
  100.                     } else {
  101.                         try {
  102.                             System.out.println("000000: 队列大小" + queue.size());
  103.                             runnable = queue.take();
  104.                             System.out.println("11111111: 队列大小" + queue.size());
  105.                         } catch (InterruptedException e) {
  106.                             e.printStackTrace();
  107.                         }
  108.                     }
  109.                 }
  110.             } finally {
  111.                 running.decrementAndGet();
  112.                 System.out.println("结束线程，当前线程数： " + running.get());
  113.             }
  114.         }
  115.     }
  116. }

8. ThreadPool 与 ForkJoinPool

参考

  1. 深入理解线程和线程池（图文详解）
  2. 通俗易懂，各常用线程池的执行 流程图
  3. (精)JAVA线程池原理以及几种线程池类型介绍
  4. 深入理解 Java 线程池：ThreadPoolExecutor
  5. 聊聊并发（三）Java线程池的分析和使用
  6. 由浅入深理解Java线程池及线程池的如何使用