ThreadPoolExecutor的详解及自定义阻塞提交的ThreadLocalExcutor实例

星期二, 2021-11-02 | Author: Lee | JAVA-and-J2EE | 2,185 views

ThreadPoolExecutor可以实现线程池的创建。ThreadPoolExecutor相关类图如下：

类图
从类图可以看出，ThreadPoolExecutor最终实现了Executor接口，是线程池创建的真正实现者。

Executor两级调度模型

Executor模型
在HotSpot虚拟机中，Java中的线程将会被一一映射为操作系统的线程。在Java虚拟机层面，用户将多个任务提交给Executor框架，Executor负责分配线程执行它们；在操作系统层面，操作系统再将这些线程分配给处理器执行。

ThreadPoolExecutor的三个角色
任务
ThreadPoolExecutor接受两种类型的任务：Callable和Runnable。

Callable：该类任务有返回结果，可以抛出异常。通过submit方法提交，返回Future对象。通过get获取执行结果。
Runnable：该类任务只执行，无法获取返回结果，在执行过程中无法抛异常。通过execute或submit方法提交。
任务执行器
Executor框架最核心的接口是Executor，它表示任务的执行器。

通过上面类图可以看出，Executor的子接口为ExecutorService。再往底层有两大实现类：ThreadPoolExecutor和ScheduledThreadPoolExecutor（集成自ThreadPoolExecutor）。

执行结果
Future接口表示异步的执行结果，它的实现类为FutureTask。

三个角色之间的处理逻辑图如下：

FutureTask逻辑
线程池处理流程

线程池处理流程
一个线程从被提交（submit）到执行共经历以下流程：

线程池判断核心线程池里是的线程是否都在执行任务，如果不是，则创建一个新的工作线程来执行任务。如果核心线程池里的线程都在执行任务，则进入下一个流程；
线程池判断工作队列是否已满。如果工作队列没有满，则将新提交的任务储存在这个工作队列里。如果工作队列满了，则进入下一个流程；
线程池判断其内部线程是否都处于工作状态。如果没有，则创建一个新的工作线程来执行任务。如果已满了，则交给饱和策略来处理这个任务。

线程池在执行execute方法时，主要有以下四种情况：

线程池执行excute方法
如果当前运行的线程少于corePoolSize，则创建新线程来执行任务（需要获得全局锁）；
如果运行的线程等于或多于corePoolSize，则将任务加入BlockingQueue；
如果无法将任务加入BlockingQueue(队列已满)，则创建新的线程来处理任务（需要获得全局锁）；
如果创建新线程将使当前运行的线程超出maxiumPoolSize，任务将被拒绝，并调用RejectedExecutionHandler.rejectedExecution()方法。
线程池采取上述的流程进行设计是为了减少获取全局锁的次数。在线程池完成预热（当前运行的线程数大于或等于corePoolSize）之后，几乎所有的excute方法调用都执行步骤二。

线程的状态流转
顺便再回顾一下线程的状态的转换，在JDK中Thread类中提供了一个枚举类，例举了线程的各个状态：

    public enum State {
 
        NEW,
 
        RUNNABLE,
 
        BLOCKED,
 
        WAITING,
 
        TIMED_WAITING,
 
        TERMINATED;
    }

一共定义了6个枚举值，其实代表的是5种类型的线程状态：

NEW：新建；
RUNNABLE：运行状态；
BLOCKED：阻塞状态；
WAITING：等待状态，WAITING和TIMED_WAITING可以归为一类，都属于等待状态，只是后者可以设置等待时间，即等待多久；
TERMINATED：终止状态；
线程关系转换图：

线程状态转换
当new Thread()说明这个线程处于NEW(新建状态)；调用Thread.start()方法表示这个线程处于RUNNABLE(运行状态)；

但是RUNNABLE状态中又包含了两种状态：READY(就绪状态)和RUNNING(运行中)。调用start()方法，线程不一定获得了CPU时间片，这时就处于READY，等待CPU时间片，当获得了CPU时间片，就处于RUNNING状态。

在运行中调用synchronized同步的代码块，没有获取到锁，这时会处于BLOCKED(阻塞状态)，当重新获取到锁时，又会变为RUNNING状态。在代码执行的过程中可能会碰到Object.wait()等一些等待方法，线程的状态又会转变为WAITING(等待状态)，等待被唤醒，当调用了Object.notifyAll()唤醒了之后线程执行完就会变为TERMINATED(终止状态)。

线程池的状态
线程池中状态通过2个二进制位（bit）来表示线程池的5个状态：RUNNING、SHUTDOWN、STOP、TIDYING和TERMINATED：

RUNNING：线程池正常工作的状态，在 RUNNING 状态下线程池接受新的任务并处理任务队列中的任务；
SHUTDOWN：调用shutdown()方法会进入 SHUTDOWN 状态。在 SHUTDOWN 状态下，线程池不接受新的任务，但是会继续执行任务队列中已有的任务；
STOP：调用shutdownNow()会进入 STOP 状态。在 STOP 状态下线程池既不接受新的任务，也不处理已经在队列中的任务。对于还在执行任务的工作线程，线程池会发起中断请求来中断正在执行的任务，同时会清空任务队列中还未被执行的任务；
TIDYING：当线程池中的所有执行任务的工作线程都已经终止，并且工作线程集合为空的时候，进入 TIDYING 状态；
TERMINATED：当线程池执行完terminated()钩子方法以后，线程池进入终态 TERMINATED；
ThreadPoolExecutor API
ThreadPoolExecutor创建线程池API：

public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler)
参数解释：

corePoolSize ：线程池常驻核心线程数。创建线程池时，线程池中并没有任何线程，当有任务来时才去创建线程，执行任务。提交一个任务，创建一个线程，直到需要执行的任务数大于线程池基本大小，则不再创建。当创建的线程数等于corePoolSize 时，会加入设置的阻塞队列。
maximumPoolSize ：线程池允许创建的最大线程数。当队列满时，会创建线程执行任务直到线程池中的数量等于maximumPoolSize。
keepAliveTime ：当线程数大于核心时，此为终止前多余的空闲线程等待新任务的最长时间。
unit ：keepAliveTime的时间单位，可选项：天（DAYS）、小时（HOURS）、分钟（MINUTES）、毫秒（MILLISECONDS）、微妙（MICROSECONDS，千分之一毫秒）和纳秒（NANOSECONDS，千分之一微妙）。
workQueue ：用来储存等待执行任务的队列。
threadFactory ：线程工厂，用来生产一组相同任务的线程。主要用于设置生成的线程名词前缀、是否为守护线程以及优先级等。设置有意义的名称前缀有利于在进行虚拟机分析时，知道线程是由哪个线程工厂创建的。
handler ：执行拒绝策略对象。当达到任务缓存上限时（即超过workQueue参数能存储的任务数），执行拒接策略。也就是当任务处理不过来的时候，线程池开始执行拒绝策略。JDK 1.5提供了四种饱和策略：
AbortPolicy：默认，直接抛异常；
只用调用者所在的线程执行任务，重试添加当前的任务，它会自动重复调用execute()方法；
DiscardOldestPolicy：丢弃任务队列中最久的任务；
DiscardPolicy：丢弃当前任务；
适当的阻塞队列
当创建的线程数等于corePoolSize，会将任务加入阻塞队列（BlockingQueue），维护着等待执行的Runnable对象。

阻塞队列通常有如下类型：

ArrayBlockingQueue ：一个由数组结构组成的有界阻塞队列。可以限定队列的长度，接收到任务时，如果没有达到corePoolSize的值，则新建线程(核心线程)执行任务，如果达到了，则入队等候，如果队列已满，则新建线程(非核心线程)执行任务，又如果总线程数到了maximumPoolSize，并且队列也满了，则发生错误。
LinkedBlockingQueue ：一个由链表结构组成的有界阻塞队列。这个队列在接收到任务时，如果当前线程数小于核心线程数，则新建线程(核心线程)处理任务；如果当前线程数等于核心线程数，则进入队列等待。由于这个队列没有最大值限制，即所有超过核心线程数的任务都将被添加到队列中，这也就导致了maximumPoolSize的设定失效，因为总线程数永远不会超过corePoolSize。
PriorityBlockingQueue ：一个支持优先级排序的无界阻塞队列。
DelayQueue：一个使用优先级队列实现的无界阻塞队列。队列内元素必须实现Delayed接口，这就意味着传入的任务必须先实现Delayed接口。这个队列在接收到任务时，首先先入队，只有达到了指定的延时时间，才会执行任务。
SynchronousQueue：一个不存储元素的阻塞队列。这个队列在接收到任务时，会直接提交给线程处理，而不保留它，如果所有线程都在工作就新建一个线程来处理这个任务。所以为了保证不出现【线程数达到了maximumPoolSize而不能新建线程】的错误，使用这个类型队列时，maximumPoolSize一般指定成Integer.MAX_VALUE，即无限大。
LinkedTransferQueue：一个由链表结构组成的无界阻塞队列。
LinkedBlockingDeque：一个由链表结构组成的双向阻塞队列。
明确的拒绝策略
当任务处理不过来时，线程池开始执行拒绝策略。

支持的拒绝策略：

ThreadPoolExecutor.AbortPolicy: 丢弃任务并抛出RejectedExecutionException异常。(默认)
ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常。
ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务。（重复此过程）
ThreadPoolExecutor.CallerRunsPolicy：由调用线程处理该任务。
线程池关闭
shutdown：将线程池状态置为SHUTDOWN，并不会立即停止。停止接收外部submit的任务，内部正在跑的任务和队列里等待的任务，会执行完后，才真正停止。
shutdownNow：将线程池状态置为STOP。企图立即停止，事实上不一定，跟shutdown()一样，先停止接收外部提交的任务，忽略队列里等待的任务，尝试将正在跑的任务interrupt中断（如果线程未处于sleep、wait、condition、定时锁状态，interrupt无法中断当前线程），返回未执行的任务列表。
awaitTermination(long timeOut, TimeUnit unit)当前线程阻塞，直到等所有已提交的任务（包括正在跑的和队列中等待的）执行完或者等超时时间到或者线程被中断，抛出InterruptedException，然后返回true（shutdown请求后所有任务执行完毕）或false（已超时）。
Executors
Executors是一个帮助类，提供了创建几种预配置线程池实例的方法：newSingleThreadExecutor、newFixedThreadPool、newCachedThreadPool等。

如果查看源码就会发现，Executors本质上就是实现了几类默认的ThreadPoolExecutor。而阿里巴巴开发手册，不建议采用Executors默认的，让使用者直接通过ThreadPoolExecutor来创建。

Executors.newSingleThreadExecutor()
创建一个单线程的线程池。这个线程池只有一个线程在工作，也就是相当于单线程串行执行所有任务。如果这个唯一的线程因为异常结束，那么会有一个新的线程来替代它。此线程池保证所有任务的执行顺序按照任务的提交顺序执行。

new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue())
该类型线程池的结构图：

newSingleThreadExecutor
该线程池的特点：

只会创建一条工作线程处理任务；
采用的阻塞队列为LinkedBlockingQueue；
Executors.newFixedThreadPool()
创建固定大小的线程池。每次提交一个任务就创建一个线程，直到线程达到线程池的最大大小。线程池的大小一旦达到最大值就会保持不变，如果某个线程因为执行异常而结束，那么线程池会补充一个新线程。

new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue());
该类型线程池的结构图：

newFixedThreadPool
该线程池的特点：

固定大小；
corePoolSize和maximunPoolSize都为用户设定的线程数量nThreads；
keepAliveTime为0，意味着一旦有多余的空闲线程，就会被立即停止掉；但这里keepAliveTime无效；
阻塞队列采用了LinkedBlockingQueue，一个无界队列；
由于阻塞队列是一个无界队列，因此永远不可能拒绝任务；
由于采用了无界队列，实际线程数量将永远维持在nThreads，因此maximumPoolSize和keepAliveTime将无效。
Executors.newCachedThreadPool()
创建一个可缓存的线程池。如果线程池的大小超过了处理任务所需要的线程，那么就会回收部分空闲（60秒不执行任务）的线程，当任务数增加时，此线程池又可以智能的添加新线程来处理任务。此线程池不会对线程池大小做限制，线程池大小完全依赖于操作系统（或者说JVM）能够创建的最大线程大小。

new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS,new SynchronousQueue());
该类型线程池的结构图：

newCachedThreadPool
该线程池的特点：

可以无限扩大；
比较适合处理执行时间比较小的任务；
corePoolSize为0，maximumPoolSize为无限大，意味着线程数量可以无限大；
keepAliveTime为60S，意味着线程空闲时间超过60s就会被杀死；
采用SynchronousQueue装等待的任务，这个阻塞队列没有存储空间，这意味着只要有请求到来，就必须要找到一条工作线程处理它，如果当前没有空闲的线程，那么就会再创建一条新的线程。
Executors.newScheduledThreadPool()
创建一个定长线程池，支持定时及周期性任务执行。

new ThreadPoolExecutor(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue());
该线程池类图：

newScheduledThreadPool
该线程池的特点：

接收SchduledFutureTask类型的任务，有两种提交任务的方式：scheduledAtFixedRate和scheduledWithFixedDelay。SchduledFutureTask接收的参数：
time：任务开始的时间
sequenceNumber：任务的序号
period：任务执行的时间间隔
采用DelayQueue存储等待的任务；
DelayQueue内部封装了一个PriorityQueue，它会根据time的先后时间排序，若time相同则根据sequenceNumber排序；
DelayQueue也是一个无界队列；
工作线程执行时，工作线程会从DelayQueue取已经到期的任务去执行；执行结束后重新设置任务的到期时间，再次放回DelayQueue；
Executors.newWorkStealingPool()
JDK8引入，创建持有足够线程的线程池支持给定的并行度，并通过使用多个队列减少竞争。

public static ExecutorService newWorkStealingPool() {
return new ForkJoinPool(Runtime.getRuntime().availableProcessors(),
ForkJoinPool.defaultForkJoinWorkerThreadFactory,
null, true);
}
Executors方法的弊端
1）newFixedThreadPool 和 newSingleThreadExecutor：允许的请求队列长度为Integer.MAX_VALUE，可能会堆积大量的请求，从而导致 OOM。2）newCachedThreadPool 和 newScheduledThreadPool：允许的创建线程数量为Integer.MAX_VALUE，可能会创建大量的线程，从而导致 OOM。

合理配置线程池大小
合理配置线程池，需要先分析任务特性，可以从以下角度来进行分析：

任务的性质：CPU密集型任务，IO密集型任务和混合型任务。
任务的优先级：高，中和低。
任务的执行时间：长，中和短。
任务的依赖性：是否依赖其他系统资源，如数据库连接。
另外，还需要查看系统的内核数：

Runtime.getRuntime().availableProcessors());
根据任务所需要的CPU和IO资源可以分为：

CPU密集型任务: 主要是执行计算任务，响应时间很快，CPU一直在运行。一般公式：线程数 = CPU核数 + 1。只有在真正的多核CPU上才能得到加速，优点是不存在线程切换开销，提高了CPU的利用率并减少了线程切换的效能损耗。
IO密集型任务：主要是进行IO操作，CPU并不是一直在执行任务，IO操作（CPU空闲状态）的时间较长，应配置尽可能多的线程，其中的线程在IO操作时，其他线程可以继续利用CPU，从而提高CPU的利用率。一般公式：线程数 = CPU核数 * 2。
使用实例
任务实现类：

public class MyThread implements Runnable{
 
   private final Integer number;
 
   public MyThread(int number){
      this.number = number;
   }
 
   public Integer getNumber() {
      return number;
   }
 
   @Override
   public void run() {
      try {
         // 业务处理
         TimeUnit.SECONDS.sleep(1);
         System.out.println("Hello! ThreadPoolExecutor - " + getNumber());
      } catch (InterruptedException e) {
         e.printStackTrace();
      }
   }
}

自定义阻塞提交的ThreadLocalExcutor：

/**
 * 自定义阻塞提交的ThreadPoolExecutor
 **/
public class CustomBlockThreadPoolExecutor {
 
   private ThreadPoolExecutor pool = null;
 
   /**
    * 线程池初始化方法
    */
   public void init() {
      // 核心线程池大小
      int poolSize = 2;
      // 最大线程池大小
      int maxPoolSize = 4;
      // 线程池中超过corePoolSize数目的空闲线程最大存活时间：30+单位TimeUnit
      long keepAliveTime = 30L;
      // ArrayBlockingQueue<Runnable> 阻塞队列容量30
      int arrayBlockingQueueSize = 30;
      pool = new ThreadPoolExecutor(poolSize, maxPoolSize, keepAliveTime,
            TimeUnit.SECONDS, new ArrayBlockingQueue<>(arrayBlockingQueueSize), new CustomThreadFactory(),
            new CustomRejectedExecutionHandler());
   }
 
   /**
    * 关闭线程池方法
    */
   public void destroy() {
      if (pool != null) {
         pool.shutdownNow();
      }
   }
 
   public ExecutorService getCustomThreadPoolExecutor() {
      return this.pool;
   }
 
   /**
    * 自定义线程工厂类，
    * 生成的线程名词前缀、是否为守护线程以及优先级等
    */
   private static class CustomThreadFactory implements ThreadFactory {
 
      private final AtomicInteger count = new AtomicInteger(0);
 
      @Override
      public Thread newThread(Runnable r) {
         Thread t = new Thread(r);
         String threadName = CustomBlockThreadPoolExecutor.class.getSimpleName() + count.addAndGet(1);
         t.setName(threadName);
         return t;
      }
   }
 
 
   /**
    * 自定义拒绝策略对象
    */
   private static class CustomRejectedExecutionHandler implements RejectedExecutionHandler {
      @Override
      public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
         // 核心改造点，将blockingqueue的offer改成put阻塞提交
         try {
            executor.getQueue().put(r);
         } catch (InterruptedException e) {
            e.printStackTrace();
         }
      }
   }
 
   /**
    * 当提交任务被拒绝时，进入拒绝机制，实现拒绝方法，把任务重新用阻塞提交方法put提交，实现阻塞提交任务功能，防止队列过大，OOM
    */
   public static void main(String[] args) {
 
      CustomBlockThreadPoolExecutor executor = new CustomBlockThreadPoolExecutor();
 
      // 初始化
      executor.init();
      ExecutorService pool = executor.getCustomThreadPoolExecutor();
      for (int i = 1; i < 51; i++) {
         MyThread myThread = new MyThread(i);
         System.out.println("提交第" + i + "个任务");
         pool.execute(myThread);
      }
 
      pool.shutdown();
      try {
         // 阻塞，超时时间到或者线程被中断
         if (!pool.awaitTermination(60, TimeUnit.SECONDS)) {
            // 立即关闭
            executor.destroy();
         }
      } catch (InterruptedException e) {
         executor.destroy();
      }
   }
}

小结
看似简单的线程池创建，其中却蕴含着各类知识，融合贯通，根据具体场景采用具体的参数进行设置才能够达到最优的效果。

总结一下就是：

用ThreadPoolExecutor自定义线程池，要看线程的用途。如果任务量不大，可以用无界队列，如果任务量非常大，要用有界队列，防止OOM；
如果任务量很大，且要求每个任务都处理成功，要对提交的任务进行阻塞提交，重写拒绝机制，改为阻塞提交。保证不抛弃一个任务；
最大线程数一般设为2N+1最好，N是CPU核数；
核心线程数，要根据任务是CPU密集型，还是IO密集型。同时，如果任务是一天跑一次，设置为0合适，因为跑完就停掉了；
如果要获取任务执行结果，用CompletionService，但是注意，获取任务的结果要重新开一个线程获取，如果在主线程获取，就要等任务都提交后才获取，就会阻塞大量任务结果，队列过大OOM，所以最好异步开个线程获取结果。

赏

Tags: thread

文章作者: Lee

本文地址： https://www.pomelolee.com/2232.html

除非注明，Pomelo Lee文章均为原创，转载请以链接形式标明本文地址

« spring.config.location启动的参数不互补可以使用spring.config.additional-location mac下的VirtualBox安装deepin自定义硬盘大小 »

No comments yet.

Search

文章分类

ajax (10)
algorithm-learn (3)
Android (6)
as (3)
computer (85)
Database (30)
disucz (4)
enterprise (1)
erlang (2)
flash (5)
golang (3)
html5 (18)
ios (4)
JAVA-and-J2EE (186)
linux (143)
mac (10)
movie-music (11)
pagemaker (36)
php (50)
spring-boot (2)
Synology群晖 (2)
Uncategorized (6)
unity (1)
webgame (15)
wordpress (33)
work-other (2)
低代码 (1)
体味生活 (40)
前端 (21)
大数据 (8)
游戏开发 (9)
爱上海 (19)
读书 (4)
软件 (3)

Pomelo Lee workspace

My workspace in Internet,Focus on Internet Technology