Java线程池及其实现原理

news/2024/10/30 9:31:10/

线程池概述

线程池（Thread Pool）是一种基于池化思想管理线程的工具，经常出现在多线程服务器中，如MySQL。

线程过多会带来额外的开销，其中包括创建销毁线程的开销、调度线程的开销等等，同时也降低了计算机的整体性能。线程池维护多个线程，等待监督管理者分配可并发执行的任务。这种做法，一方面避免了处理任务时创建销毁线程开销的代价，另一方面避免了线程数量膨胀导致的过分调度问题，保证了对内核的充分利用。

使用线程池可以带来的一系列好处：

  • 降低资源消耗：通过池化技术重复利用已创建的线程，降低线程创建和销毁造成的损耗。
  • 提高响应速度：任务到达时，无需等待线程创建即可立即执行。
  • 提高线程的可管理性：线程是稀缺资源，如果无限制创建，不仅会消耗系统资源，还会因为线程的不合理分布导致资源调度失衡，降低系统的稳定性。使用线程池可以进行统一的分配、调优和监控。
  • 提供更多更强大的功能：线程池具备可拓展性，允许开发人员向其中增加更多的功能。比如延时定时线程池ScheduledThreadPoolExecutor，就允许任务延期执行或定期执行。

线程池解决的问题

线程池解决的核心问题就是资源管理问题。在并发环境下，系统不能够确定在任意时刻中，有多少任务需要执行，有多少资源需要投入。这种不确定性将带来以下若干问题：

  1. 频繁申请/销毁资源和调度资源，将带来额外的消耗，可能会非常巨大。
  2. 对资源无限申请缺少抑制手段，易引发系统资源耗尽的风险。
  3. 系统无法合理管理内部的资源分布，会降低系统的稳定性。

为解决资源分配这个问题，线程池采用了“池化”（Pooling）思想。为了最大化收益并最小化风险，而将资源统一在一起管理的一种思想。

“池化”思想不仅仅能应用在计算机领域，在金融、设备、人员管理、工作管理等领域也有相关的应用。

在计算机领域中的表现为：统一管理IT资源，包括服务器、存储、和网络资源等等。通过共享资源，使用户在低投入中获益。除去线程池，还有其他比较典型的几种使用策略包括：

  1. 内存池(Memory Pooling)：预先申请内存，提升申请内存速度，减少内存碎片。
  2. 连接池(Connection Pooling)：预先申请数据库连接，提升申请连接的速度，降低系统的开销。
  3. 对象实例池(Object Pooling)：循环使用对象，减少资源在初始化和释放时的昂贵损耗。

线程池核心设计与实现

JDK 1.8线程池ThreadPoolExecutor的UML类图
  • 顶层接口Executor提供了一种思想：将任务提交和任务执行进行解耦。用户无需关注如何创建线程，如何调度线程来执行任务，用户只需提供Runnable对象，将任务的运行逻辑提交到执行器(Executor)中，由Executor框架完成线程的调配和任务的执行部分。
  • ExecutorService接口增加了一些能力：
  1. 扩充执行任务的能力，补充可以为一个或一批异步任务生成Future的方法；
  2. 提供了管控线程池的方法，比如停止线程池的运行。
  • AbstractExecutorService则是上层的抽象类，将执行任务的流程串联了起来，保证下层的实现只需关注一个执行任务的方法即可。
  • 最下层的实现类ThreadPoolExecutor实现了顶层接口Executor。ThreadPoolExecutor实现最复杂的运行部分，ThreadPoolExecutor将会一方面维护自身的生命周期，另一方面同时管理线程和任务，使两者良好的结合从而执行并行任务。

ThreadPoolExecutor运行机制

ThreadPoolExecutor运行流程

线程池在内部实际上构建了一个生产者消费者模型，将线程和任务两者解耦，并不直接关联，从而良好的缓冲任务，复用线程。

线程池的运行主要分成两部分：任务管理、线程管理。

任务管理部分充当生产者的角色，当任务提交后，线程池会判断该任务后续的流转：

  1. 直接申请线程执行该任务；
  2. 缓冲到队列中等待线程执行；
  3. 拒绝该任务。

线程管理部分是消费者，它们被统一维护在线程池内，根据任务请求进行线程的分配，当线程执行完任务后则会继续获取新的任务去执行，最终当线程获取不到任务的时候，线程就会被回收。

生命周期管理

线程池运行的状态，并不是用户显式设置的，而是伴随着线程池的运行，由内部来维护。线程池内部使用一个变量维护两个值：运行状态(runState)和线程数量 (workerCount)。在具体实现中，线程池将运行状态(runState)、线程数量 (workerCount)两个关键参数的维护放在了一起，如下代码所示：

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
private static final int COUNT_BITS = Integer.SIZE - 3;
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

ctl这个AtomicInteger类型变量，是对线程池的运行状态和线程池中有效线程的数量进行控制的一个字段， 它同时包含两部分的信息：线程池的运行状态 (runState) 和线程池内有效线程的数量 (workerCount)，高3位保存runState，低29位保存workerCount，两个变量之间互不干扰。用一个变量去存储两个值，可避免在做相关决策时，出现不一致的情况，不必为了维护两者的一致，而占用锁资源。通过阅读线程池源代码也可以发现，经常出现要同时判断线程池运行状态和线程数量的情况。线程池也提供了若干方法去供用户获得线程池当前的运行状态、线程个数。这里都使用的是位运算的方式，相比于基本运算，速度也会快很多。

关于内部封装的获取生命周期状态、获取线程池线程数量的计算方法如以下代码所示：

private static int runStateOf(int c)     { return c & ~CAPACITY; } //计算当前运行状态
private static int workerCountOf(int c)  { return c & CAPACITY; }  //计算当前线程数量
private static int ctlOf(int rs, int wc) { return rs | wc; }   //通过状态和线程数生成ctl

五种运行状态

ThreadPoolExecutor的运行状态有5种，分别为：

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;
运行状态状态描述
RUNNING

能接受新提交的任务，并且也能处理阻塞队列中的任务。

SHUTDOWN

关闭状态。不再接受新提交的任务，但却可以继续处理阻塞队列中已保存的任务。

STOP

不能接受新任务，也不处理队列中的惹怒我，会中断正在处理的线程。

TIDYING

所有的任务都已终止，workerCount（有效线程数）为0

TERMINATED

在terminated()方法执行完后进入该状态

原子锁

private final ReentrantLock mainLock = new ReentrantLock();
private final Condition termination = mainLock.newCondition();

生命周期转换

线程池生命周期

Executor 框架的使用

  1. 主线程首先创建实现 Runnable或Callable 接口的任务对象。
  2. 把创建完成的事项 Runnable/Callable 接口的对象直接交给ExecutorService执行。ExecutorService.execute(Runnable command) 或者把 Runnable对象或Callable对象提交给ExecutorService执行 
    ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
cachedThreadPool.execute(Runnable command);
cachedThreadPool.submit(Runnable task);
cachedThreadPool.submit(Callable<T> task);
  3. 如果执行ExecutorService execute(...)，ExecutorService 将返回一个事项Future接口的对象。由于FutureTask实现了Runnable，也可以在使用时创建FutureTask。然后交给ExecutorService执行。
  4. 最后，主线程可以执行FutureTask.get()方法来等待任务执行完成。主线程也可以执行FutureTask.cancel(boolean mayInterruptIfRunning)来取消此任务的执行。

 核心数据结构

public class ThreadPoolExecutor extends AbstractExecutorService {//...private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));// 存放任务的阻塞队列private final BlockingQueue<Runnable> workQueue;// 对线程池内部各种变量进行互斥访问控制private final ReentrantLock mainLock = new ReentrantLock();// 线程集合private final HashSet<Worker> workers = new HashSet<Worker>();//...
}

每一个线程是一个Worker对象。Worker是ThreadPoolExector的内部类，核心数据结构如下：

private final class Worker extends AbstractQueuedSynchronizer implements Runnable {// ...// Worker封装的线程final Thread thread;// Worker接收到的第1个任务Runnable firstTask;// Worker执行完毕的任务个数volatile long completedTasks;// ...
}

 Worker继承于AQS，也就是说Worker本身就是一把锁。 这把锁用于线程池的关闭、线程执行任务的过程中。

七大核心参数

参数说明
corePoolSize核心线程数量，线程池维护线程的最少数量
maximumPoolSize线程池维护线程的最大数量
keepAliveTime线程池除核心线程外的其他线程的最长空闲时间，超过该时间的空闲线程会被销毁
unitkeepAliveTime的单位，TimeUnit中的几个静态属性：NANOSECONDS、MICROSECONDS、MILLISECONDS、SECONDS
workQueue线程池所使用的任务缓冲队列
threadFactory线程工厂，用于创建线程，一般用默认的即可
handler线程池对拒绝任务的处理策略
public ThreadPoolExecutor(int corePoolSize, // 核心线程数int maximumPoolSize, // 最大线程数long keepAliveTime, // 当线程数大于核心线程数时，多余的空闲线程存活最长时间TimeUnit unit, // 空闲线程存活时间的时间单位BlockingQueue<Runnable> workQueue, // 任务队列ThreadFactory threadFactory, // 线程工厂，用来创建线程RejectedExecutionHandler handler) { // 拒绝策略this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,Executors.defaultThreadFactory(), defaultHandler);
}

线程池的处理流程

线程池的处理流程
  1. 如果此时线程池中的数量小于corePoolSize，即使线程池中的线程都处于空闲状态，也要创建新的线程来处理被添加的任务。
  2. 如果此时线程池中的数量等于corePoolSize，但是缓冲队列workQueue未满，那么任务被放入缓冲队列。
  3. 如果此时线程池中的数量大于等于corePoolSize，缓冲队列workQueue满，并且线程池中的数量小于maximumPoolSize，建新的线程来处理被添加的任务。
  4. 如果此时线程池中的数量大于corePoolSize，缓冲队列workQueue满，并且线程池中的数量等于maximumPoolSize，那么通过 handler所指定的策略来处理此任务。
  5. 当线程池中的线程数量大于 corePoolSize时，如果某线程空闲时间超过keepAliveTime，线程将被终止。这样，线程池可以动态的调整池中的线程数。

总结：处理任务判断的优先级为 核心线程corePoolSize、任务队列workQueue、最大线程maximumPoolSize，如果三者都满了，使用handler处理被拒绝的任务。

注意：

  1. 当workQueue使用的是无界限队列时，maximumPoolSize参数就变的无意义了，比如new LinkedBlockingQueue(),或者new ArrayBlockingQueue(Integer.MAX_VALUE)。
  2. 使用SynchronousQueue队列时由于该队列没有容量的特性，所以不会对任务进行排队，如果线程池中没有空闲线程，会立即创建一个新线程来接收这个任务。maximumPoolSize要设置大一点。
  3. 核心线程和最大线程数量相等时keepAliveTime无作用。

线程池关闭

  1. shutdown() 不接收新任务,会处理已添加任务
  2. shutdownNow() 不接受新任务,不处理已添加任务,中断正在处理的任务

常用线程池

Executors

.newCachedThreadPool();

创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。
内部实现：new ThreadPoolExecutor(0,Integer.MAX_VALUE,60L,

TimeUnit.SECONDS,new SynchronousQueue());

Executors

.newFixedThreadPool(int);

创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。
内部实现：new ThreadPoolExecutor(nThreads, nThreads,0L,TimeUnit.MILLISECONDS,new LinkedBlockingQueue());

Executors

.newSingleThreadExecutor();

创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照顺序执行。
内部实现：new ThreadPoolExecutor(1,1,0L,TimeUnit.MILLISECONDS,new LinkedBlockingQueue())

Executors

.newScheduledThreadPool(int)

创建一个定长线程池，支持定时及周期性任务执行。
内部实现：new ScheduledThreadPoolExecutor(corePoolSize)
Executors
.newSingleThreadScheduledExecutor()
创建一个单线程的任务调度池(定时任务/延时任务)

队列

名称描述
ArrayBlockingQueue一个用数组实现的有界阻塞队列，此队列按照先进先出(FIFO)的原则对元素进行排序。支持公平锁和非公平锁。
LinkedBlockingQueue一个由链表结构组成的有界队列，此队列按照先进先出FIFO)的原则对元素进行排序。此队列的默认长度为Integer.MAXVALUE，所以默认创建的该队列有容量危险。
PriorityBlockingQueue一个支持线程优先级排序的无界队列，默认自然序进行排序，也可以自定义实现compareTo(方法来指定元素排字规则，不能保证同优先级元素的顺序。
DelayQueue一个实现PriorityBlockingQueue实现延迟获取的无界队列，在创建元素时，可以指定多久才能从队列中获取当前元素。只有延时期满后才能从队列中获取元素。
SynchronousQueue一个不存储元素的阻塞队列，每一个put操作必须等待take操作，否则不能添加元素。支持公平锁和非公平锁SynchronousQueue的-个使用场景是在线程池里。Executors.newCachedThreadPool0就使用了SynchronousQueue，这个线程池根据需要(新任务到来时)创建新的线程，如果有空闲线程则会重复使用线程空闲了60秒后会被回收。
LinkedTransferQueue一个由链表结构组成的无界阳塞队列，相当于其它队列，LinkedTransferQueue队列多了transfer和trvTransfer方法。
LinkedBlockingDeque一个由链表结构组成的双向阻塞队列。队列头部和尾部都可以添加和移除元素，多线程并发时，可以将锁的竞争最多降到一半。

拒绝策略

任务拒绝模块是线程池的保护部分，线程池有一个最大的容量，当线程池的任务缓存队列已满，并且线程池中的线程数目达到maximumPoolSize时，就需要拒绝掉该任务，采取任务拒绝策略，保护线程池。

拒绝策略是一个接口，其设计如下：

public interface RejectedExecutionHandler {void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
}

 四种拒绝策略

名称描述
ThreadPoolExecutor.AbortPolicy丢弃任务并抛出RejectedExecutionException异常。 这是线程池默认的拒绝策略，在任务不能再提交的时候，抛出异常，及时反馈程序运行状态。如果是比较关键的业务推荐使用此拒绝策略，这样子在系统不能承载更大的并发量的时候，能够及时的通过异常发现。
ThreadPoolExecutor.DiscardPolicy丢弃任务，但是不抛出异常。 使用此策略，可能会使我们无法发现系统的异常状态。建议是一些无关紧要的业务采用此策略。
ThreadPoolExecutor.DiscardOldestPolicy丢弃队列最前面的任务，然后重新提交被拒绝的任务。是否要采用此种拒绝策略，还得根据实际业务是否允许丢弃老任务来认真衡量。
ThreadPoolExecutor.CallerRunsPolicy由调用线程(提交任务的线程)处理该任务。这种情况是需要让所有任务都执行完毕那么就适合大量计算的任务类型去执行，多线程仅仅是增大吞吐量的手段，最终必须要让每个任务都执行完毕。

Worker线程管理

Worker线程

线程池为了掌握线程的状态并维护线程的生命周期，设计了线程池内的工作线程Worker。

private final class Worker extends AbstractQueuedSynchronizer implements Runnable{final Thread thread;//Worker持有的线程Runnable firstTask;//初始化的任务，可以为null
}

Worker执行任务的模型

Worker执行任务

 线程池需要管理线程的生命周期，需要在线程长时间不运行的时候进行回收。线程池使用一张Hash表去持有线程的引用，这样可以通过添加引用、移除引用这样的操作来控制线程的生命周期。这个时候重要的就是如何判断线程是否在运行。

​Worker是通过继承AQS，使用AQS来实现独占锁这个功能。没有使用可重入锁ReentrantLock，而是使用AQS，为的就是实现不可重入的特性去反应线程现在的执行状态。

  1. lock方法一旦获取了独占锁，表示当前线程正在执行任务中。
  2. 如果正在执行任务，则不应该中断线程。
  3. 如果该线程现在不是独占锁的状态，也就是空闲的状态，说明它没有在处理任务，这时可以对该线程进行中断。
  4. 线程池在执行shutdown方法或tryTerminate方法时会调用interruptIdleWorkers方法来中断空闲的线程，interruptIdleWorkers方法会使用tryLock方法来判断线程池中的线程是否是空闲状态；如果线程是空闲状态则可以安全回收。

在线程回收过程中就使用到了这种特性，回收过程如下图所示：

线程池回收过程

线程池使用场景

场景1：快速响应用户请求

场景2：快速处理批量任务

【附】阿里巴巴Java开发手册中对线程池的使用规范

  1. 【强制】创建线程或线程池时请指定有意义的线程名称，方便出错时回溯
  2. 【强制】线程资源必须通过线程池提供，不允许在应用中自行显式创建线程。
    说明： 使用线程池的好处是减少在创建和销毁线程上所花的时间以及系统资源的开销，解决资源不足的问题。如果不使用线程池，有可能造成系统创建大量同类线程而导致消耗完内存或者“过度切换”的问题。
  3. 【强制】线程池不允许使用 Executors 去创建，而是通过 ThreadPoolExecutor 的方式创建，这样的处理方式让写的同学更加明确线程池的运行规则，规避资源耗尽的风险。
    说明： Executors 返回的线程池对象的弊端如下：
    1） FixedThreadPool 和 SingleThreadPool:
    允许的请求队列长度为 Integer.MAX_VALUE，可能会堆积大量的请求，从而导致 OOM。
    2） CachedThreadPool 和 ScheduledThreadPool:
    允许的创建线程数量为 Integer.MAX_VALUE， 可能会创建大量的线程，从而导致 OOM。

http://www.ppmy.cn/news/62877.html

相关文章

Vue3 watch 监听对象数组中对象的特定属性

Vue3 watch 监听对象数组中对象的特定属性

在 Vue 3 中&#xff0c;可以使用 watch 函数来监听对象数组中对象的特定属性。可以通过在回调函数中遍历数组来检查对象的特定属性是否发生变化&#xff0c;并在变化发生时执行相应的操作。 一、监听对象的特定属性 例如&#xff0c;假设有一个名为 items 的对象数组&#x…
阅读更多...
欧科云链OKLink：2023年4月安全事件盘点

欧科云链OKLink：2023年4月安全事件盘点

一、基本信息 2023年4月安全事件共造约6000万美金的损失&#xff0c;与上个月相比&#xff0c;损失金额有所降落&#xff0c;但安全事件数量依旧不减。其中&#xff0c;Yearn Finance因参数配置错误&#xff0c;导致了1000多万美金的损失。同时&#xff0c;有一些已经出现过的…
阅读更多...
platform总线五级匹配解析

platform总线五级匹配解析

代码来源&#xff1a;开源linux内核linux-6.2.9 platform总线设备与驱动的匹配 对于device和driver无论哪个创建都会尝试主动寻找对方进行绑定&#xff0c;而platform bus总线的匹配原则如上面的代码所示&#xff0c;共有五级匹配&#xff0c;这里进行详细解析下&#xff1a; …
阅读更多...
CSAPP第九章 虚拟内存

CSAPP第九章 虚拟内存

概述 程序是存放在虚拟地址空间中&#xff0c;运行是在物理空间中&#xff0c;如何进行转换是虚拟内存需要研究的问题。 完成逻辑地址到虚拟地址再从虚拟地址到逻辑地址映射的机构叫MMU 两个部件 段式管理和页式管理 之前有提到&#xff0c;cache的管理是通过cache映射表&…
阅读更多...
SIM800C模块AT指令测试（三）语音功能相关

SIM800C模块AT指令测试（三）语音功能相关

1. 紧急呼叫操作 每个国家区域的紧急号码不同&#xff0c;客户可通过AT指令设置需要的号码&#xff0c;最多可以设置11个。 ATCEMNL? CEMNL: (0-1),(1-11),("0"-"999")... OK ATCEMNL1,3,“112”,“000”,“911” // 设置紧急号码 OK AT…
阅读更多...
块状链表实现BigString大字符串操作（golang）

块状链表实现BigString大字符串操作（golang）

前言 块状链表是介于链表和数组之间的数据结构&#xff0c;能够在 O ( n ) O(\sqrt{n}) O(n ​)时间内完成插入、删除、访问操作。 数据结构如图所示。假设最大容量为 n n n, 则它有一个长度为 s n s\sqrt{n} sn ​的链表。链表中每个结点是一个长度为 2 n 2 \times \sqrt{…
阅读更多...
配置本地Angular环境并使用VsCode调试Angular前端项目

配置本地Angular环境并使用VsCode调试Angular前端项目

配置本地Angular环境并使用VsCode调试Angular前端项目 配置本地Angular环境部署Node.Js本地环境配置一下环境变量 使用vscode调试Angular安装vscode 配置本地Angular环境 部署Node.Js本地环境 1 从官网下载node.js, 本文为(v16.13.0) 下载地址: https://nodejs.org/dist/v16.…
阅读更多...
硬件设计 之 PCIe常用知识

硬件设计 之 PCIe常用知识

以下是本人在自己在设计PCIe中常遇到的一些知识&#xff0c;对他们进行了简单整理一下&#xff0c;包括基本定义、传输速率、layout要求等。比如作为硬件工程师要了解芯片架构&#xff0c;哪些PCIe接口可以使用&#xff0c;使用这些PCIe要做什么&#xff0c;需要使用PCIe x1还是…
阅读更多...
最新文章

Copyright @ 2022~2023