Java多线程之线程池

1. ThreadPool线程池

1.1 线程池的使用

• 线程复用、控制最大并发数、管理线程

  • 降低资源消耗。通过重复利用已创建的线程降低线程创建和销毁造成的销耗。
  • 提高响应速度。当任务到达时，任务可以不需要等待线程创建就能立即执行。
  • 提高线程的可管理性。线程是稀缺资源，如果无限制的创建，不仅会销耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一的分配，调优和监控。

• 线程池架构

  

  ​ Executor，Executors，ExecutorService，ThreadPoolExecutor

1.1.1 三大方法

• Executors.newFixedThreadPool(int)

  • 执行长期任务性能好，创建一个线程池，一池有N个固定的线程，有固定线程数的线程

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

  • newFixedThreadPool创建的线程池corePoolSize和maximumPoolSize值是相等的，它使用的是LinkedBlockingQueue

• Executors.newSingleThreadExecutor()

  • 一个任务一个任务的执行，一池一线程

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

  • newSingleThreadExecutor 创建的线程池corePoolSize和maximumPoolSize值都是1，它使用的是LinkedBlockingQueue

• Executors.newCachedThreadPool()

  • 执行很多短期异步任务，线程池根据需要创建新线程，但在先前构建的线程可用时将重用它们。可扩容，遇强则强

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

  • newCachedThreadPool创建的线程池将corePoolSize设置为0，将maximumPoolSize设置为Integer.MAX_VALUE，它使用的是SynchronousQueue，也就是说来了任务就创建线程运行，当线程空闲超过60秒，就销毁线程。

• 从以上源码可以看出，三大方法底层均是使用ThreadPoolExecutor()来创建线程池

• 代码

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 线程池
 * Arrays
 * Collections
 * Executors
 */
public class MyThreadPoolDemo {

    public static void main(String[] args) {
        //List list = new ArrayList();
        //List list = Arrays.asList("a","b");
        //固定数的线程池，一池五线程
	//一个银行网点，5个受理业务的窗口
	//ExecutorService threadPool =  Executors.newFixedThreadPool(5); 
        //一个银行网点，1个受理业务的窗口
	//ExecutorService threadPool =  Executors.newSingleThreadExecutor(); 
        //一个银行网点，可扩展受理业务的窗口
        ExecutorService threadPool =  Executors.newCachedThreadPool(); 

        //10个顾客请求
        try {
            for (int i = 1; i <=10; i++) {
                threadPool.execute(()->{
                    System.out.println(Thread.currentThread().getName()+"	 办理业务");
                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            threadPool.shutdown();
        }
    }
}

• 工作中均不使用以上三大方法来创建线程池，而是直接使用ThreadPoolExecutor()来创建线程池

  

public static void main(String[] args) {
    ExecutorService threadPool = new ThreadPoolExecutor(
        2,
        5,
        2L,
        TimeUnit.SECONDS,
        new ArrayBlockingQueue<Runnable>(3),
        Executors.defaultThreadFactory(),
        //new ThreadPoolExecutor.AbortPolicy()
        //new ThreadPoolExecutor.CallerRunsPolicy()
        //new ThreadPoolExecutor.DiscardOldestPolicy()
        new ThreadPoolExecutor.DiscardOldestPolicy()
    );
    //10个顾客请求
    try {
        for (int i = 1; i <= 10; i++) {
            threadPool.execute(() -> {
                System.out.println(Thread.currentThread().getName() + "	 办理业务");
            });
        }
    } catch (Exception e) {
        e.printStackTrace();
    } finally {
        threadPool.shutdown();
    }
}

1.2 七大参数

• ThreadPoolExecutor()源码

/**
 * Creates a new {@code ThreadPoolExecutor} with the given initial
 * parameters.
 *
 * @param corePoolSize the number of threads to keep in the pool, even
 *        if they are idle, unless {@code allowCoreThreadTimeOut} is set
 * @param maximumPoolSize the maximum number of threads to allow in the
 *        pool
 * @param keepAliveTime when the number of threads is greater than
 *        the core, this is the maximum time that excess idle threads
 *        will wait for new tasks before terminating.
 * @param unit the time unit for the {@code keepAliveTime} argument
 * @param workQueue the queue to use for holding tasks before they are
 *        executed.  This queue will hold only the {@code Runnable}
 *        tasks submitted by the {@code execute} method.
 * @param threadFactory the factory to use when the executor
 *        creates a new thread
 * @param handler the handler to use when execution is blocked
 *        because the thread bounds and queue capacities are reached
 * @throws IllegalArgumentException if one of the following holds:<br>
 *         {@code corePoolSize < 0}<br>
 *         {@code keepAliveTime < 0}<br>
 *         {@code maximumPoolSize <= 0}<br>
 *         {@code maximumPoolSize < corePoolSize}
 * @throws NullPointerException if {@code workQueue}
 *         or {@code threadFactory} or {@code handler} is null
 */
public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) {
    if (corePoolSize < 0 ||
        maximumPoolSize <= 0 ||
        maximumPoolSize < corePoolSize ||
        keepAliveTime < 0)
        throw new IllegalArgumentException();
    if (workQueue == null || threadFactory == null || handler == null)
        throw new NullPointerException();
    this.acc = System.getSecurityManager() == null ?
        null :
    AccessController.getContext();
    this.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.workQueue = workQueue;
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    this.threadFactory = threadFactory;
    this.handler = handler;
}

• 七大参数
  • corePoolSize：线程池中的常驻核心线程数
  • maximumPoolSize：线程池中能够容纳同时执行的最大线程数，此值必须大于等于1
  • keepAliveTime：多余的空闲线程的存活时间，当前池中线程数量超过corePoolSize时，当空闲时间达到keepAliveTime时，多余线程会被销毁直到只剩下corePoolSize个线程为止
  • unit：keepAliveTime的单位
  • workQueue：任务队列，被提交但尚未被执行的任务
  • threadFactory：表示生成线程池中工作线程的线程工厂，用于创建线程，一般默认的即可
  • handler：拒绝策略，表示当队列满了，并且工作线程大于等于线程池的最大线程数（maximumPoolSize）时如何来拒绝请求执行的runnable的策略

1.3 四种拒绝策略

• 拒绝策略：等待队列已经排满了，再也塞不下新任务了，同时，线程池中的max线程也达到了，无法继续为新任务服务。这个是时候我们就需要拒绝策略机制合理的处理这个问题。

• JDK内置四种拒绝策略

  

  • AbortPolicy(默认)：直接抛出RejectedExecutionException异常阻止系统正常运行
  • CallerRunsPolicy：“调用者运行”一种调节机制，该策略既不会抛弃任务，也不会抛出异常，而是将某些任务回退到调用者，从而降低新任务的流量。
  • DiscardOldestPolicy：抛弃队列中等待最久的任务，然后把当前任务加人队列中尝试再次提交当前任务。
  • DiscardPolicy：该策略默默地丢弃无法处理的任务，不予任何处理也不抛出异常。如果允许任务丢失，这是最好的一种策略。

  /**
  * new ThreadPoolExecutor.AbortPolicy() // 银行满了，还有人进来，不处理这个人的，抛出异
  常
  * new ThreadPoolExecutor.CallerRunsPolicy() // 哪来的去哪里！
  * new ThreadPoolExecutor.DiscardPolicy() //队列满了，丢掉任务，不会抛出异常！
  * new ThreadPoolExecutor.DiscardOldestPolicy() //队列满了，尝试去和最早的竞争，也不会抛出异常！
  */
  

• 以上内置拒绝策略均实现了RejectedExecutionHandle接口

1.4 线程池底层运行原理

• 在创建了线程池后，开始等待请求。

• 当调用execute()方法添加一个请求任务时，线程池会做出如下判断：

  • 1如果正在运行的线程数量小于corePoolSize，那么马上创建线程运行这个任务；
  • 2如果正在运行的线程数量大于或等于corePoolSize，那么将这个任务放入队列；
  • 3如果这个时候队列满了且正在运行的线程数量还小于maximumPoolSize，那么还是要创建非核心线程立刻运行这个任务(队列中的依旧在等待)；
  • 4如果队列满了且正在运行的线程数量大于或等于maximumPoolSize，那么线程池会启动饱和拒绝策略来执行。

• 当一个线程完成任务时，它会从队列中取下一个任务来执行。

• 当一个线程无事可做超过一定的时间（keepAliveTime）时，线程会判断：

  • 如果当前运行的线程数大于corePoolSize，那么这个线程就被停掉。所以线程池的所有任务完成后，它最终会收缩到corePoolSize的大小。

• 举例：

  • 银行有5个窗口(maximumPoolSize)，2个启用(corePoolSize)，3个暂停服务，且等待区有5张椅子供等待使用(workQueue)，开始时前面进来的2个客户直接到启用的2个窗口办理业务，后面来的客户，先到等待区椅子上等待，当等待区5张椅子坐满后，又有人进来办业务，于是银行就启用另外3个窗口进行服务，办理完业务的窗口，直接喊等待区的人去那个窗口办理，当5个窗口都在服务，且等待区也满的时候，银行只能让保安在门口堵着(RejectedExecutionHandler)，拒绝后面的人员进入(毕竟疫情期间不能挤在一起嘛！！！)

  import java.util.concurrent.*;
  
  /**
   * @Description TODO
   * @Package: juc.juc
   * @ClassName ThreadPoolDemo
   * @author: wuwb
   * @date: 2020/12/18 9:12
   */
  public class ThreadPoolDemo {
      public static void main(String[] args) {
          ExecutorService threadPool = new ThreadPoolExecutor(
                  2,
                  5,
                  2L,
                  TimeUnit.SECONDS,
                  new ArrayBlockingQueue<Runnable>(5),
                  Executors.defaultThreadFactory(),
                  new ThreadPoolExecutor.CallerRunsPolicy()
          );
          try {
              for (int i = 1; i <= 10; i++) {
                  final int j = i;
                  threadPool.execute(()->{
                      System.out.println(Thread.currentThread().getName() + " run " + j);
                      try {
                          TimeUnit.SECONDS.sleep(2);
                      } catch (InterruptedException e) {
                          e.printStackTrace();
                      }
                  });
              }
          } catch (Exception e) {
              e.printStackTrace();
          } finally {
              threadPool.shutdown();
          }
      }
  }
  /*
  pool-1-thread-1 run 1
  pool-1-thread-2 run 2
  pool-1-thread-3 run 8
  pool-1-thread-4 run 9
  pool-1-thread-5 run 10
  pool-1-thread-1 run 3
  pool-1-thread-4 run 4
  pool-1-thread-2 run 5
  pool-1-thread-3 run 6
  pool-1-thread-5 run 7
  
  1、2进核心线程，3、4、5、6、7进队列等待，8、9、10启用非核心线程先于队列中任务运行
  */
  

  *CPU密集型 最大线程数为CPU核数，CPU核数Runtime.getRuntime().availableProcessors()；