zoukankan      html  css  js  c++  java
  • Java多线程之ThreadPoolExecutor详解使用

    1、概述

    我将讲解JAVA原生线程池的基本使用,并由此延伸出JAVA中和线程管理相关的类结构体系,然后我们详细描述JAVA原生线程池的结构和工作方式

    2、为什么要使用线程池

    这里写图片描述

    前文我们已经讲到,线程是一个操作系统概念。操作系统负责这个线程的创建、挂起、运行、阻塞和终结操作。而操作系统创建线程、切换线程状态、终结线程都要进行CPU调度。

    另一方面,目前大多数生产环境我们所面临问题的技术背景一般是:处理某一次请求的时间是非常短暂的,但是请求数量是巨大的。这种技术背景下,如果我们为每一个请求都单独创建一个线程,那么物理机的所有资源基本上都被操作系统创建线程、切换线程状态、销毁线程这些操作所占用,用于业务请求处理的资源反而减少了。所以最理想的处理方式是,将处理请求的线程数量控制在一个范围,既保证后续的请求不会等待太长时间,又保证物理机将足够的资源用于请求处理本身。

    另外,一些操作系统是有最大线程数量限制的。当运行的线程数量逼近这个值的时候,操作系统会变得不稳定。这也是我们要限制线程数量的原因。

    3、线程池的基本使用方式

    JAVA语言为我们提供了两种基础线程池的选择:ScheduledThreadPoolExecutor和ThreadPoolExecutor。它们都实现了ExecutorService接口(注意,ExecutorService接口本身和“线程池”并没有直接关系,它的定义更接近“执行器”,而“使用线程管理的方式进行实现”只是其中的一种实现方式)。这篇文章中,我们主要围绕ThreadPoolExecutor类进行讲解。

    3-1、简单使用

    首先我们来看看ThreadPoolExecutor类的最简单使用方式:

    1.  
      package test.thread.pool;
    2.  
       
    3.  
      import java.util.concurrent.SynchronousQueue;
    4.  
      import java.util.concurrent.ThreadPoolExecutor;
    5.  
      import java.util.concurrent.TimeUnit;
    6.  
       
    7.  
      import org.apache.commons.logging.Log;
    8.  
      import org.apache.commons.logging.LogFactory;
    9.  
      import org.apache.log4j.BasicConfigurator;
    10.  
       
    11.  
      public class PoolThreadSimple {
    12.  
       
    13.  
      static {
    14.  
      BasicConfigurator.configure();
    15.  
      }
    16.  
       
    17.  
      public static void main(String[] args) throws Throwable {
    18.  
       
    19.  
      /*
    20.  
      * corePoolSize:核心大小,线程池初始化的时候,就会有这么大
    21.  
      * maximumPoolSize:线程池最大线程数
    22.  
      * keepAliveTime:如果当前线程池中线程数大于corePoolSize。
    23.  
      * 多余的线程,在等待keepAliveTime时间后如果还没有新的线程任务指派给它,它就会被回收
    24.  
      *
    25.  
      * unit:等待时间keepAliveTime的单位
    26.  
      *
    27.  
      * workQueue:等待队列。这个对象的设置是本文将重点介绍的内容
    28.  
      * */
    29.  
      ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(5, 10, 1, TimeUnit.MINUTES, new SynchronousQueue<Runnable>());
    30.  
      for(int index = 0 ; index < 10 ; index ++) {
    31.  
      poolExecutor.submit(new PoolThreadSimple.TestRunnable(index));
    32.  
      }
    33.  
       
    34.  
      // 没有特殊含义,只是为了保证main线程不会退出
    35.  
      synchronized (poolExecutor) {
    36.  
      poolExecutor.wait();
    37.  
      }
    38.  
      }
    39.  
       
    40.  
      /**
    41.  
      * 这个就是测试用的线程
    42.  
      * @author yinwenjie
    43.  
      */
    44.  
      private static class TestRunnable implements Runnable {
    45.  
       
    46.  
      /**
    47.  
      * 日志
    48.  
      */
    49.  
      private static Log LOGGER = LogFactory.getLog(TestRunnable.class);
    50.  
       
    51.  
      /**
    52.  
      * 记录任务的唯一编号,这样在日志中好做识别
    53.  
      */
    54.  
      private Integer index;
    55.  
       
    56.  
      public TestRunnable(int index) {
    57.  
      this.index = index;
    58.  
      }
    59.  
       
    60.  
      /**
    61.  
      * @return the index
    62.  
      */
    63.  
      public Integer getIndex() {
    64.  
      return index;
    65.  
      }
    66.  
       
    67.  
      @Override
    68.  
      public void run() {
    69.  
      /*
    70.  
      * 线程中,就只做一件事情:
    71.  
      * 等待60秒钟的事件,以便模拟业务操作过程
    72.  
      * */
    73.  
      Thread currentThread = Thread.currentThread();
    74.  
      TestRunnable.LOGGER.info("线程:" + currentThread.getId() + " 中的任务(" + this.getIndex() + ")开始执行===");
    75.  
      synchronized (currentThread) {
    76.  
      try {
    77.  
      currentThread.wait(60000);
    78.  
      } catch (InterruptedException e) {
    79.  
      TestRunnable.LOGGER.error(e.getMessage(), e);
    80.  
      }
    81.  
      }
    82.  
       
    83.  
      TestRunnable.LOGGER.info("线程:" + currentThread.getId() + " 中的任务(" + this.getIndex() + ")执行完成");
    84.  
      }
    85.  
       
    86.  
      }
    87.  
      }

    随后的文章中我们将对线程池中的corePoolSize、maximumPoolSize、keepAliveTime、timeUnit、workQueue、threadFactory、handler参数和一些常用/不常用的设置项进行逐一讲解。

    3-2、ThreadPoolExecutor逻辑结构和工作方式

    在上面的代码中,我们创建线程池的时候使用了ThreadPoolExecutor中最简单的一个构造函数:

    1.  
      public ThreadPoolExecutor(int corePoolSize,
    2.  
      int maximumPoolSize,
    3.  
      long keepAliveTime,
    4.  
      TimeUnit unit,
    5.  
      BlockingQueue<Runnable> workQueue)
    • 构造函数中需要传入的参数包括corePoolSize、maximumPoolSize、keepAliveTime、timeUnit和workQueue。要明确理解这些参数(和后续将要介绍的参数)的含义,就首先要搞清楚ThreadPoolExecutor线程池的逻辑结构。

    这里写图片描述

    一定要注意一个概念,即存在于线程池中容器的一定是Thread对象,而不是您要求运行的任务(所以叫线程池而不叫任务池也不叫对象池,更不叫游泳池);您要求运行的任务将被线程池分配给某一个空闲的Thread运行。

    从上图中,我们可以看到构成线程池的几个重要元素:

    • 等待队列:顾名思义,就是您调用线程池对象的submit()方法或者execute()方法,要求线程池运行的任务(这些任务必须实现Runnable接口或者Callable接口)。但是出于某些原因线程池并没有马上运行这些任务,而是送入一个队列等待执行(这些原因后文马上讲解)。

    • 核心线程:线程池主要用于执行任务的是“核心线程”,“核心线程”的数量是您创建线程时所设置的corePoolSize参数决定的。如果不进行特别的设定,线程池中始终会保持corePoolSize数量的线程数(不包括创建阶段)。

    • 非核心线程:一旦任务数量过多(由等待队列的特性决定),线程池将创建“非核心线程”临时帮助运行任务。您设置的大于corePoolSize参数小于maximumPoolSize参数的部分,就是线程池可以临时创建的“非核心线程”的最大数量。这种情况下如果某个线程没有运行任何任务,在等待keepAliveTime时间后,这个线程将会被销毁,直到线程池的线程数量重新达到corePoolSize。

    • 要重点理解上一条描述中黑体字部分的内容。也就是说,并不是所谓的“非核心线程”才会被回收;而是谁的空闲时间达到keepAliveTime这个阀值,就会被回收。直到线程池中线程数量等于corePoolSize为止。

    • maximumPoolSize参数也是当前线程池允许创建的最大线程数量。那么如果您设置的corePoolSize参数和您设置的maximumPoolSize参数一致时,线程池在任何情况下都不会回收空闲线程。keepAliveTime和timeUnit也就失去了意义。

    • keepAliveTime参数和timeUnit参数也是配合使用的。keepAliveTime参数指明等待时间的量化值,timeUnit指明量化值单位。例如keepAliveTime=1,timeUnit为TimeUnit.MINUTES,代表空闲线程的回收阀值为1分钟。

    说完了线程池的逻辑结构,下面我们讨论一下线程池是怎样处理某一个运行任务的。下图描述了一个完整的任务处理过程:

    这里写图片描述

    1、首先您可以通过线程池提供的submit()方法或者execute()方法,要求线程池执行某个任务。线程池收到这个要求执行的任务后,会有几种处理情况:

    1.1、如果当前线程池中运行的线程数量还没有达到corePoolSize大小时,线程池会创建一个新的线程运行您的任务,无论之前已经创建的线程是否处于空闲状态。

    1.2、如果当前线程池中运行的线程数量已经达到设置的corePoolSize大小,线程池会把您的这个任务加入到等待队列中。直到某一个的线程空闲了,线程池会根据您设置的等待队列规则,从队列中取出一个新的任务执行。

    1.3、如果根据队列规则,这个任务无法加入等待队列。这时线程池就会创建一个“非核心线程”直接运行这个任务。注意,如果这种情况下任务执行成功,那么当前线程池中的线程数量一定大于corePoolSize。

    1.4、如果这个任务,无法被“核心线程”直接执行,又无法加入等待队列,又无法创建“非核心线程”直接执行,且您没有为线程池设置RejectedExecutionHandler。这时线程池会抛出RejectedExecutionException异常,即线程池拒绝接受这个任务。(实际上抛出RejectedExecutionException异常的操作,是ThreadPoolExecutor线程池中一个默认的RejectedExecutionHandler实现:AbortPolicy,这在后文会提到)

    2、一旦线程池中某个线程完成了任务的执行,它就会试图到任务等待队列中拿去下一个等待任务(所有的等待任务都实现了BlockingQueue接口,按照接口字面上的理解,这是一个可阻塞的队列接口),它会调用等待队列的poll()方法,并停留在哪里。

    3、当线程池中的线程超过您设置的corePoolSize参数,说明当前线程池中有所谓的“非核心线程”。那么当某个线程处理完任务后,如果等待keepAliveTime时间后仍然没有新的任务分配给它,那么这个线程将会被回收。线程池回收线程时,对所谓的“核心线程”和“非核心线程”是一视同仁的,直到线程池中线程的数量等于您设置的corePoolSize参数时,回收过程才会停止。

    3-3、不常用的设置

    在ThreadPoolExecutor线程池中,有一些不常用的设置。我建议如果您在应用场景中没有特殊的要求,就不需要使用这些设置:

    3-3-1、 allowCoreThreadTimeOut:

    前文我们讨论到,线程池回收线程只会发生在当前线程池中线程数量大于corePoolSize参数的时候;当线程池中线程数量小于等于corePoolSize参数的时候,回收过程就会停止。

    allowCoreThreadTimeOut设置项可以要求线程池:将包括“核心线程”在内的,没有任务分配的任何线程,在等待keepAliveTime时间后全部进行回收:

    1.  
      ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(5, 10, 1, TimeUnit.MINUTES, new ArrayBlockingQueue<Runnable>(1));
    2.  
       
    3.  
      poolExecutor.allowCoreThreadTimeOut(true);

    以下是设置前的效果:

    这里写图片描述

    以下是设置后的效果:

    这里写图片描述

    3-3-2 prestartAllCoreThreads

    前文我们还讨论到,当线程池中的线程还没有达到您设置的corePoolSize参数值的时候,如果有新的任务到来,线程池将创建新的线程运行这个任务,无论之前已经创建的线程是否处于空闲状态。这个描述可以用下面的示意图表示出来:

    这里写图片描述

    prestartAllCoreThreads设置项,可以在线程池创建,但还没有接收到任何任务的情况下,先行创建符合corePoolSize参数值的线程数:

    1、概述

    我将讲解JAVA原生线程池的基本使用,并由此延伸出JAVA中和线程管理相关的类结构体系,然后我们详细描述JAVA原生线程池的结构和工作方式

    2、为什么要使用线程池

    这里写图片描述

    前文我们已经讲到,线程是一个操作系统概念。操作系统负责这个线程的创建、挂起、运行、阻塞和终结操作。而操作系统创建线程、切换线程状态、终结线程都要进行CPU调度。

    另一方面,目前大多数生产环境我们所面临问题的技术背景一般是:处理某一次请求的时间是非常短暂的,但是请求数量是巨大的。这种技术背景下,如果我们为每一个请求都单独创建一个线程,那么物理机的所有资源基本上都被操作系统创建线程、切换线程状态、销毁线程这些操作所占用,用于业务请求处理的资源反而减少了。所以最理想的处理方式是,将处理请求的线程数量控制在一个范围,既保证后续的请求不会等待太长时间,又保证物理机将足够的资源用于请求处理本身。

    另外,一些操作系统是有最大线程数量限制的。当运行的线程数量逼近这个值的时候,操作系统会变得不稳定。这也是我们要限制线程数量的原因。

    3、线程池的基本使用方式

    JAVA语言为我们提供了两种基础线程池的选择:ScheduledThreadPoolExecutor和ThreadPoolExecutor。它们都实现了ExecutorService接口(注意,ExecutorService接口本身和“线程池”并没有直接关系,它的定义更接近“执行器”,而“使用线程管理的方式进行实现”只是其中的一种实现方式)。这篇文章中,我们主要围绕ThreadPoolExecutor类进行讲解。

    3-1、简单使用

    首先我们来看看ThreadPoolExecutor类的最简单使用方式:

    1.  
      package test.thread.pool;
    2.  
       
    3.  
      import java.util.concurrent.SynchronousQueue;
    4.  
      import java.util.concurrent.ThreadPoolExecutor;
    5.  
      import java.util.concurrent.TimeUnit;
    6.  
       
    7.  
      import org.apache.commons.logging.Log;
    8.  
      import org.apache.commons.logging.LogFactory;
    9.  
      import org.apache.log4j.BasicConfigurator;
    10.  
       
    11.  
      public class PoolThreadSimple {
    12.  
       
    13.  
      static {
    14.  
      BasicConfigurator.configure();
    15.  
      }
    16.  
       
    17.  
      public static void main(String[] args) throws Throwable {
    18.  
       
    19.  
      /*
    20.  
      * corePoolSize:核心大小,线程池初始化的时候,就会有这么大
    21.  
      * maximumPoolSize:线程池最大线程数
    22.  
      * keepAliveTime:如果当前线程池中线程数大于corePoolSize。
    23.  
      * 多余的线程,在等待keepAliveTime时间后如果还没有新的线程任务指派给它,它就会被回收
    24.  
      *
    25.  
      * unit:等待时间keepAliveTime的单位
    26.  
      *
    27.  
      * workQueue:等待队列。这个对象的设置是本文将重点介绍的内容
    28.  
      * */
    29.  
      ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(5, 10, 1, TimeUnit.MINUTES, new SynchronousQueue<Runnable>());
    30.  
      for(int index = 0 ; index < 10 ; index ++) {
    31.  
      poolExecutor.submit(new PoolThreadSimple.TestRunnable(index));
    32.  
      }
    33.  
       
    34.  
      // 没有特殊含义,只是为了保证main线程不会退出
    35.  
      synchronized (poolExecutor) {
    36.  
      poolExecutor.wait();
    37.  
      }
    38.  
      }
    39.  
       
    40.  
      /**
    41.  
      * 这个就是测试用的线程
    42.  
      * @author yinwenjie
    43.  
      */
    44.  
      private static class TestRunnable implements Runnable {
    45.  
       
    46.  
      /**
    47.  
      * 日志
    48.  
      */
    49.  
      private static Log LOGGER = LogFactory.getLog(TestRunnable.class);
    50.  
       
    51.  
      /**
    52.  
      * 记录任务的唯一编号,这样在日志中好做识别
    53.  
      */
    54.  
      private Integer index;
    55.  
       
    56.  
      public TestRunnable(int index) {
    57.  
      this.index = index;
    58.  
      }
    59.  
       
    60.  
      /**
    61.  
      * @return the index
    62.  
      */
    63.  
      public Integer getIndex() {
    64.  
      return index;
    65.  
      }
    66.  
       
    67.  
      @Override
    68.  
      public void run() {
    69.  
      /*
    70.  
      * 线程中,就只做一件事情:
    71.  
      * 等待60秒钟的事件,以便模拟业务操作过程
    72.  
      * */
    73.  
      Thread currentThread = Thread.currentThread();
    74.  
      TestRunnable.LOGGER.info("线程:" + currentThread.getId() + " 中的任务(" + this.getIndex() + ")开始执行===");
    75.  
      synchronized (currentThread) {
    76.  
      try {
    77.  
      currentThread.wait(60000);
    78.  
      } catch (InterruptedException e) {
    79.  
      TestRunnable.LOGGER.error(e.getMessage(), e);
    80.  
      }
    81.  
      }
    82.  
       
    83.  
      TestRunnable.LOGGER.info("线程:" + currentThread.getId() + " 中的任务(" + this.getIndex() + ")执行完成");
    84.  
      }
    85.  
       
    86.  
      }
    87.  
      }

    随后的文章中我们将对线程池中的corePoolSize、maximumPoolSize、keepAliveTime、timeUnit、workQueue、threadFactory、handler参数和一些常用/不常用的设置项进行逐一讲解。

    3-2、ThreadPoolExecutor逻辑结构和工作方式

    在上面的代码中,我们创建线程池的时候使用了ThreadPoolExecutor中最简单的一个构造函数:

    1.  
      public ThreadPoolExecutor(int corePoolSize,
    2.  
      int maximumPoolSize,
    3.  
      long keepAliveTime,
    4.  
      TimeUnit unit,
    5.  
      BlockingQueue<Runnable> workQueue)
    • 构造函数中需要传入的参数包括corePoolSize、maximumPoolSize、keepAliveTime、timeUnit和workQueue。要明确理解这些参数(和后续将要介绍的参数)的含义,就首先要搞清楚ThreadPoolExecutor线程池的逻辑结构。

    这里写图片描述

    一定要注意一个概念,即存在于线程池中容器的一定是Thread对象,而不是您要求运行的任务(所以叫线程池而不叫任务池也不叫对象池,更不叫游泳池);您要求运行的任务将被线程池分配给某一个空闲的Thread运行。

    从上图中,我们可以看到构成线程池的几个重要元素:

    • 等待队列:顾名思义,就是您调用线程池对象的submit()方法或者execute()方法,要求线程池运行的任务(这些任务必须实现Runnable接口或者Callable接口)。但是出于某些原因线程池并没有马上运行这些任务,而是送入一个队列等待执行(这些原因后文马上讲解)。

    • 核心线程:线程池主要用于执行任务的是“核心线程”,“核心线程”的数量是您创建线程时所设置的corePoolSize参数决定的。如果不进行特别的设定,线程池中始终会保持corePoolSize数量的线程数(不包括创建阶段)。

    • 非核心线程:一旦任务数量过多(由等待队列的特性决定),线程池将创建“非核心线程”临时帮助运行任务。您设置的大于corePoolSize参数小于maximumPoolSize参数的部分,就是线程池可以临时创建的“非核心线程”的最大数量。这种情况下如果某个线程没有运行任何任务,在等待keepAliveTime时间后,这个线程将会被销毁,直到线程池的线程数量重新达到corePoolSize。

    • 要重点理解上一条描述中黑体字部分的内容。也就是说,并不是所谓的“非核心线程”才会被回收;而是谁的空闲时间达到keepAliveTime这个阀值,就会被回收。直到线程池中线程数量等于corePoolSize为止。

    • maximumPoolSize参数也是当前线程池允许创建的最大线程数量。那么如果您设置的corePoolSize参数和您设置的maximumPoolSize参数一致时,线程池在任何情况下都不会回收空闲线程。keepAliveTime和timeUnit也就失去了意义。

    • keepAliveTime参数和timeUnit参数也是配合使用的。keepAliveTime参数指明等待时间的量化值,timeUnit指明量化值单位。例如keepAliveTime=1,timeUnit为TimeUnit.MINUTES,代表空闲线程的回收阀值为1分钟。

    说完了线程池的逻辑结构,下面我们讨论一下线程池是怎样处理某一个运行任务的。下图描述了一个完整的任务处理过程:

    这里写图片描述

    1、首先您可以通过线程池提供的submit()方法或者execute()方法,要求线程池执行某个任务。线程池收到这个要求执行的任务后,会有几种处理情况:

    1.1、如果当前线程池中运行的线程数量还没有达到corePoolSize大小时,线程池会创建一个新的线程运行您的任务,无论之前已经创建的线程是否处于空闲状态。

    1.2、如果当前线程池中运行的线程数量已经达到设置的corePoolSize大小,线程池会把您的这个任务加入到等待队列中。直到某一个的线程空闲了,线程池会根据您设置的等待队列规则,从队列中取出一个新的任务执行。

    1.3、如果根据队列规则,这个任务无法加入等待队列。这时线程池就会创建一个“非核心线程”直接运行这个任务。注意,如果这种情况下任务执行成功,那么当前线程池中的线程数量一定大于corePoolSize。

    1.4、如果这个任务,无法被“核心线程”直接执行,又无法加入等待队列,又无法创建“非核心线程”直接执行,且您没有为线程池设置RejectedExecutionHandler。这时线程池会抛出RejectedExecutionException异常,即线程池拒绝接受这个任务。(实际上抛出RejectedExecutionException异常的操作,是ThreadPoolExecutor线程池中一个默认的RejectedExecutionHandler实现:AbortPolicy,这在后文会提到)

    2、一旦线程池中某个线程完成了任务的执行,它就会试图到任务等待队列中拿去下一个等待任务(所有的等待任务都实现了BlockingQueue接口,按照接口字面上的理解,这是一个可阻塞的队列接口),它会调用等待队列的poll()方法,并停留在哪里。

    3、当线程池中的线程超过您设置的corePoolSize参数,说明当前线程池中有所谓的“非核心线程”。那么当某个线程处理完任务后,如果等待keepAliveTime时间后仍然没有新的任务分配给它,那么这个线程将会被回收。线程池回收线程时,对所谓的“核心线程”和“非核心线程”是一视同仁的,直到线程池中线程的数量等于您设置的corePoolSize参数时,回收过程才会停止。

    3-3、不常用的设置

    在ThreadPoolExecutor线程池中,有一些不常用的设置。我建议如果您在应用场景中没有特殊的要求,就不需要使用这些设置:

    3-3-1、 allowCoreThreadTimeOut:

    前文我们讨论到,线程池回收线程只会发生在当前线程池中线程数量大于corePoolSize参数的时候;当线程池中线程数量小于等于corePoolSize参数的时候,回收过程就会停止。

    allowCoreThreadTimeOut设置项可以要求线程池:将包括“核心线程”在内的,没有任务分配的任何线程,在等待keepAliveTime时间后全部进行回收:

    1.  
      ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(5, 10, 1, TimeUnit.MINUTES, new ArrayBlockingQueue<Runnable>(1));
    2.  
       
    3.  
      poolExecutor.allowCoreThreadTimeOut(true);
    • 1
    • 2
    • 3

    以下是设置前的效果:

    这里写图片描述

    以下是设置后的效果:

    这里写图片描述

    3-3-2 prestartAllCoreThreads

    前文我们还讨论到,当线程池中的线程还没有达到您设置的corePoolSize参数值的时候,如果有新的任务到来,线程池将创建新的线程运行这个任务,无论之前已经创建的线程是否处于空闲状态。这个描述可以用下面的示意图表示出来:

    这里写图片描述

    prestartAllCoreThreads设置项,可以在线程池创建,但还没有接收到任何任务的情况下,先行创建符合corePoolSize参数值的线程数:

  • 相关阅读:
    如何在js中使用递归
    基于angular写的一个todolist
    使用github参与开源项目
    用sass写栅格系统
    Activity返回按钮
    Listview优化MovieListAdapter的使用
    [强悍]listview下拉刷新,上拉加载更多组件版
    Google自己的下拉刷新组件SwipeRefreshLayout
    当ListView有Header时,onItemClick里的position不正确
    tabhost练习,剥离自“去哪儿”
  • 原文地址:https://www.cnblogs.com/dim2046/p/12059073.html
Copyright © 2011-2022 走看看