线程池原理--执行器ThreadPoolExecutor

2024-06-15 00:58
文章标签 线程 原理 threadpoolexecutor 执行器

本文主要是介绍线程池原理--执行器ThreadPoolExecutor，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

文章目录

  • 线程池原理--执行器ThreadPoolExecutor
    • 属性
    • 构造器
      • 构造器参数介绍
    • execute()方法

线程池原理–总索引

线程池原理–执行器ThreadPoolExecutor

ThreadPoolExecutor 是Executor的核心实现类。

属性

  • 线程池运行状态
    • RUNNING:接受新的任务，并且处理队列中的任务
    • SHUTDOWN: 不接受新的任务，但是仍然处理队列中的任务
    • STOP: 不接受新的任务，也不处理队列中的任务
    • TIDYING: 所有的任务已经结束, workerCount 为0,程序会调用钩子方法
      terminated()，这个什么也没做。
    • TERMINATED: 所有的任务都已经完成。
private static final int RUNNING    = -1 << COUNT_BITS;
private static final int SHUTDOWN   =  0 << COUNT_BITS;
private static final int STOP       =  1 << COUNT_BITS;
private static final int TIDYING    =  2 << COUNT_BITS;
private static final int TERMINATED =  3 << COUNT_BITS;

在这里插入图片描述

  • AtomicInteger ctl
    原子整形包装的线程池控制状态(The main pool control state)。
    在这里插入图片描述
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
//rs :高3位，runState , wc : 低29位，workerCount，线程池中当前活动的线程数量
private static int ctlOf(int rs, int wc) { return rs | wc; }
  • CAPACITY 线程池的最大容量
private static final int COUNT_BITS = Integer.SIZE - 3;
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;
  • 任务队列，用于存放执行未结束的任务。详细看线程池原理–任务队列BlockingQueue
private final BlockingQueue<Runnable> workQueue;
  • ReentrantLock 用于对于工作集和related bookkeeping（不知道啥意思）的并发访问锁控制。
private final ReentrantLock mainLock = new ReentrantLock();
  • 工作线程集，Worker是ThreadPoolExecutor类的内部类，并实现了Runnable接口，用户提交的任务都是给Worker线程执行。
private final HashSet<Worker> workers = new HashSet<>();
  • Tracks largest attained pool size. Accessed only under mainLock.
private int largestPoolSize;

private final Condition termination = mainLock.newCondition();
  • 完成任务的计数器，当任务结束时更新。
private long completedTaskCount;
  • 创建新线程的工厂
private volatile ThreadFactory threadFactory;
  • 拒绝策略，详情察看
    线程池原理–拒绝策略之RejectedExecutionHandler类
private volatile RejectedExecutionHandler handler;
  • 默认的拒绝策略
private static final RejectedExecutionHandler defaultHandler = new AbortPolicy();
  • 当 allowCoreThreadTimeOut设置为true.
    如果当前线程池中的线程如果大于corePoolSize，那么如果空闲时间超过keepAliveTime，那么就会销毁掉一些线程。
    否则，就一直等到有新的任务执行。
 private volatile long keepAliveTime;
  • true:核心线程即使空闲也不会被销毁掉。
  • false:核心线程空闲超过keepAliveTime定义的超时时间，则会被销毁掉。
private volatile boolean allowCoreThreadTimeOut;
  • 线程池核心线程大小
private volatile int corePoolSize;
  • 最大的线程池大小 = 核心线程大小 + 非核心线程大小
private volatile int maximumPoolSize;

  • 运行时权限

 private static final RuntimePermission shutdownPerm =new RuntimePermission("modifyThread");
  • 调用finalize会用到
 private final AccessControlContext acc;this.acc = System.getSecurityManager() == null ?null :AccessController.getContext();

构造器

 public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue) {this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,Executors.defaultThreadFactory(), defaultHandler);}
public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory) {this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,threadFactory, defaultHandler);}
public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,RejectedExecutionHandler handler) {this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,Executors.defaultThreadFactory(), handler);}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;}

构造器参数介绍

下面来解释下各个参数：

  • int corePoolSize：该线程池中核心线程数最大值
    核心线程：线程池新建线程的时候，如果当前线程总数小于corePoolSize，则新建的是核心线程，如果超过corePoolSize，则新建的是非核心线程,核心线程默认情况下会一直存活在线程池中，即使这个核心线程啥也不干(闲置状态)。
    如果指定ThreadPoolExecutor的allowCoreThreadTimeOut这个属性为true，那么核心线程如果不干活(闲置状态)的话，超过一定时间(时长下面参数决定)，就会被销毁掉。

  • int maximumPoolSize： 该线程池中线程总数最大值
    线程总数 = 核心线程数 + 非核心线程数。

  • long keepAliveTime：
    当 allowCoreThreadTimeOut设置为true.
    如果当前线程池中的线程如果大于corePoolSize，那么如果空闲时间超过keepAliveTime，那么就会销毁掉一些线程。
    否则，就一直等到有新的任务执行。

  • TimeUnit unit：keepAliveTime的单位
    TimeUnit是一个枚举类型，其包括：
    NANOSECONDS ： 1微毫秒 = 1微秒 / 1000
    MICROSECONDS ： 1微秒 = 1毫秒 / 1000
    MILLISECONDS ： 1毫秒 = 1秒 /1000
    SECONDS ： 秒
    MINUTES ： 分
    HOURS ： 小时
    DAYS ： 天

  • BlockingQueue ArrayBlockingQueue ：一个由数组结构组成的有界阻塞队列。

    • LinkedBlockingQueue ：一个由链表结构组成的有界阻塞队列。
    • PriorityBlockingQueue ：一个支持优先级排序的无界阻塞队列。
    • DelayQueue： 一个使用优先级队列实现的无界阻塞队列。
    • SynchronousQueue： 一个不存储元素的阻塞队列。
    • LinkedTransferQueue： 一个由链表结构组成的无界阻塞队列。
    • LinkedBlockingDeque： 一个由链表结构组成的双向阻塞队列。

  • ThreadFactory threadFactory：线程工厂，用于创建线程执行 我们提交的任务。

  • RejectedExecutionHandler handler：这个指定当队列满时继续添加任务该u做如何处理，详细可以看线程池原理–拒绝策略之RejectedExecutionHandler类。

execute()方法

execute()方法用于提交任务。

//提交的Runnable接口的实现类
//实际用户可以提交Runnable接口的实现类或者Callable接口的实现类，AbstractExecutorService会在submit()方法中进行预处理，将Callable类型对象转化为Runnable类型对象
public void execute(Runnable command) {if (command == null)throw new NullPointerException();         *//** Proceed in 3 steps:** 1. If fewer than corePoolSize threads are running, try to* start a new thread with the given command as its first* task.  The call to addWorker atomically checks runState and* workerCount, and so prevents false alarms that would add* threads when it shouldn't, by returning false.** 2. If a task can be successfully queued, then we still need* to double-check whether we should have added a thread* (because existing ones died since last checking) or that* the pool shut down since entry into this method. So we* recheck state and if necessary roll back the enqueuing if* stopped, or start a new thread if there are none.** 3. If we cannot queue task, then we try to add a new* thread.  If it fails, we know we are shut down or saturated* and so reject the task.*/int c = ctl.get();if (workerCountOf(c) < corePoolSize) {if (addWorker(command, true))return;c = ctl.get();}if (isRunning(c) && workQueue.offer(command)) {int recheck = ctl.get();if (! isRunning(recheck) && remove(command))reject(command);else if (workerCountOf(recheck) == 0)addWorker(null, false);}else if (!addWorker(command, false))reject(command);}

在这里插入图片描述

  1. 如果当前运行的线程少于corePoolSize，则创建新线程来执行任务（需要获得全局锁）
  2. 如果运行的线程等于或多于corePoolSize ,则将任务加入BlockingQueue
  3. 如果无法将任务加入BlockingQueue(队列已满)，则创建新的线程来处理任务（需要获得全局锁）
  4. 如果创建新线程将使当前运行的线程超出maxiumPoolSize，任务将被拒绝，并调用RejectedExecutionHandler.rejectedExecution()方法。
    在这里插入图片描述
  • addWorker(Runnable firstTask, boolean core)

private boolean addWorker(Runnable firstTask, boolean core) {retry:for (;;) {int c = ctl.get();int rs = runStateOf(c);// Check if queue empty only if necessary.if (rs >= SHUTDOWN &&! (rs == SHUTDOWN &&firstTask == null &&! workQueue.isEmpty()))return false;for (;;) {int wc = workerCountOf(c);if (wc >= CAPACITY ||wc >= (core ? corePoolSize : maximumPoolSize))return false;if (compareAndIncrementWorkerCount(c))break retry;c = ctl.get();  // Re-read ctlif (runStateOf(c) != rs)continue retry;// else CAS failed due to workerCount change; retry inner loop}}boolean workerStarted = false;boolean workerAdded = false;Worker w = null;try {w = new Worker(firstTask);final Thread t = w.thread;if (t != null) {final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {// Recheck while holding lock.// Back out on ThreadFactory failure or if// shut down before lock acquired.int rs = runStateOf(ctl.get());if (rs < SHUTDOWN ||(rs == SHUTDOWN && firstTask == null)) {if (t.isAlive()) // precheck that t is startablethrow new IllegalThreadStateException();workers.add(w);int s = workers.size();if (s > largestPoolSize)largestPoolSize = s;workerAdded = true;}} finally {mainLock.unlock();}if (workerAdded) {t.start();workerStarted = true;}}} finally {if (! workerStarted)addWorkerFailed(w);}return workerStarted;}

参数：
firstTask： worker线程的初始任务，可以为空
core： true：将corePoolSize作为上限，false：将maximumPoolSize作为上限
addWorker方法有4种传参的方式：

1、addWorker(command, true)2、addWorker(command, false)3、addWorker(null, false)4、addWorker(null, true)

在execute方法中就使用了前3种，结合这个核心方法进行以下分析
第一个：线程数小于corePoolSize时，放一个需要处理的task进Workers Set。如果Workers Set长度超过corePoolSize，就返回false
第二个：当队列被放满时，就尝试将这个新来的task直接放入Workers Set，而此时Workers Set的长度限制是maximumPoolSize。如果线程池也满了的话就返回false
第三个：放入一个空的task进workers Set，长度限制是maximumPoolSize。这样一个task为空的worker在线程执行的时候会去任务队列里拿任务，这样就相当于创建了一个新的线程，只是没有马上分配任务
第四个：这个方法就是放一个null的task进Workers Set，而且是在小于corePoolSize时，如果此时Set中的数量已经达到corePoolSize那就返回false，什么也不干。实际使用中是在prestartAllCoreThreads()方法，这个方法用来为线程池预先启动corePoolSize个worker等待从workQueue中获取任务执行

这篇关于线程池原理--执行器ThreadPoolExecutor的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/1061982。

相关文章

Spring Security 单点登录与自动登录机制的实现原理

Spring Security 单点登录与自动登录机制的实现原理

《SpringSecurity单点登录与自动登录机制的实现原理》本文探讨SpringSecurity实现单点登录（SSO）与自动登录机制,涵盖JWT跨系统认证、RememberMe持久化Token... 目录一、核心概念解析1.1 单点登录（SSO）1.2 自动登录（Remember Me）二、代码分析三、
阅读更多...
在MySQL中实现冷热数据分离的方法及使用场景底层原理解析

在MySQL中实现冷热数据分离的方法及使用场景底层原理解析

《在MySQL中实现冷热数据分离的方法及使用场景底层原理解析》MySQL冷热数据分离通过分表/分区策略、数据归档和索引优化,将频繁访问的热数据与冷数据分开存储,提升查询效率并降低存储成本,适用于高并发... 目录实现冷热数据分离1. 分表策略2. 使用分区表3. 数据归档与迁移在mysql中实现冷热数据分
阅读更多...
Java中的xxl-job调度器线程池工作机制

Java中的xxl-job调度器线程池工作机制

《Java中的xxl-job调度器线程池工作机制》xxl-job通过快慢线程池分离短时与长时任务,动态降级超时任务至慢池,结合异步触发和资源隔离机制,提升高频调度的性能与稳定性,支撑高并发场景下的可靠... 目录⚙️ 一、调度器线程池的核心设计 二、线程池的工作流程 三、线程池配置参数与优化 四、总结：线程
阅读更多...
WinForm跨线程访问UI及UI卡死的解决方案

WinForm跨线程访问UI及UI卡死的解决方案

《WinForm跨线程访问UI及UI卡死的解决方案》在WinForm开发过程中,跨线程访问UI控件和界面卡死是常见的技术难题,由于Windows窗体应用程序的UI控件默认只能在主线程（UI线程）上操作... 目录前言正文案例1：直接线程操作（无UI访问）案例2：BeginInvoke访问UI（错误用法）案例
阅读更多...
Linux线程之线程的创建、属性、回收、退出、取消方式

Linux线程之线程的创建、属性、回收、退出、取消方式

《Linux线程之线程的创建、属性、回收、退出、取消方式》文章总结了线程管理核心知识：线程号唯一、创建方式、属性设置（如分离状态与栈大小）、回收机制（join/detach）、退出方法（返回/pthr... 目录1. 线程号2. 线程的创建3. 线程属性4. 线程的回收5. 线程的退出6. 线程的取消7.
阅读更多...
Linux下进程的CPU配置与线程绑定过程

Linux下进程的CPU配置与线程绑定过程

《Linux下进程的CPU配置与线程绑定过程》本文介绍Linux系统中基于进程和线程的CPU配置方法,通过taskset命令和pthread库调整亲和力,将进程/线程绑定到特定CPU核心以优化资源分配... 目录1 基于进程的CPU配置1.1 对CPU亲和力的配置1.2 绑定进程到指定CPU核上运行2 基于
阅读更多...
Javaee多线程之进程和线程之间的区别和联系(最新整理)

Javaee多线程之进程和线程之间的区别和联系(最新整理)

《Javaee多线程之进程和线程之间的区别和联系(最新整理)》进程是资源分配单位,线程是调度执行单位,共享资源更高效,创建线程五种方式:继承Thread、Runnable接口、匿名类、lambda,r... 目录进程和线程进程线程进程和线程的区别创建线程的五种写法继承Thread,重写run实现Runnab
阅读更多...
SpringBoot线程池配置使用示例详解

SpringBoot线程池配置使用示例详解

《SpringBoot线程池配置使用示例详解》SpringBoot集成@Async注解,支持线程池参数配置（核心数、队列容量、拒绝策略等）及生命周期管理,结合监控与任务装饰器,提升异步处理效率与系统... 目录一、核心特性二、添加依赖三、参数详解四、配置线程池五、应用实践代码说明拒绝策略（Rejected
阅读更多...
Java 线程安全与 volatile与单例模式问题及解决方案

Java 线程安全与 volatile与单例模式问题及解决方案

《Java线程安全与volatile与单例模式问题及解决方案》文章主要讲解线程安全问题的五个成因（调度随机、变量修改、非原子操作、内存可见性、指令重排序）及解决方案,强调使用volatile关键字... 目录什么是线程安全线程安全问题的产生与解决方案线程的调度是随机的多个线程对同一个变量进行修改线程的修改操
阅读更多...
从原理到实战深入理解Java 断言assert

从原理到实战深入理解Java 断言assert

《从原理到实战深入理解Java断言assert》本文深入解析Java断言机制,涵盖语法、工作原理、启用方式及与异常的区别,推荐用于开发阶段的条件检查与状态验证,并强调生产环境应使用参数验证工具类替代... 目录深入理解 Java 断言（assert）：从原理到实战引言：为什么需要断言？一、断言基础1.1 语
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2