多线程-线程池

server/2025/3/6 6:20:44/

简介

线程池：池化技术的一种，用于存储线程，实现线程的复用。线程池机制避免了频繁的创建和销毁线程，可以提升程序的性能

线程池的使用

核心API：ThreadPoolExecutor

ThreadPoolExecutor是线程池的核心类，提供了线程池的核心功能

构造方法：

java">public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler)

构建线程池的7个核心参数：

  • 核心线程数：corePoolSize，线程池中保持的线程数量，即使在空闲状态下。
  • 最大线程数：maximumPoolSize，线程池中最多能创建多少个线程
  • 线程存活时间：keepAliveTime，当线程池中的线程数大于核心线程数时，如果一个线程空闲的时间达到最大存活时间，则会终止，直到线程池中的线程数不超过核心线程数
  • 线程存活时间的单位：unit，参数keepAliveTime的时间单位，有7种取值，是java.util.concurrent.TimeUnit类中枚举的7个常量值
  • 阻塞队列：workQueue，用来存储等待执行的任务
  • 线程工厂：threadFactory，主要用来创建线程，设置线程的名称等特性
  • 拒绝策略：handler，无法处理任务时的策略

线程池的工作机制

  • 当线程数目小于核心线程数时：每来一个任务，就会创建一个线程去执行这个任务
  • 当线程数目等于核心线程数但是阻塞队列没有满时：把提交的任务添加到阻塞队列
  • 当阻塞队列已满但是没有到最大线程数时：创建新的线程执行任务，直到最大线程数。
  • 当阻塞队列已满并且达到最大线程数时：执行拒绝策略，此时线程池已经无法执行任务
  • 当线程空闲时：如果线程池中的线程数量大于核心线程数时，当某线程空闲时间超过最大存活时间，线程将被终止，直至线程池中的线程数目不大于核心线程数

常见的阻塞队列

  • 基于数组的有界阻塞队列：ArrayBlockingQueue，队列按先进先出原则对元素进行排序，创建时需要指定队列长度
  • 基于链表的无界阻塞队列：LinkedBlockingQueue，同样是先进先出，因为队列无界，所以核心线程数满了之后，不会再创建线程了。使用无界队列需要预防任务堆积导致的OOM
  • 基于优先级的无界阻塞队列：PriorityBlockingQueue，用户可以指定元素的排序规则
  • 不存储元素的阻塞队列：SynchronousQueue，当线程池不需阻塞存队列时，可以使用它。向队列中新增元素和从队列中取元素两个操作是匹配的，插入元素时会检查是否有等待取元素的线程，如果没有，会阻塞，直到有线程来取元素。取元素时也是一样，会先检查有没有新增元素的线程，如果有，取元素，否则阻塞。

常见的拒绝策略

拒绝策略是指当线程池无法执行任务时该怎么办，线程池提供的拒绝策略：

  • 丢弃任务并抛异常：这是默认的拒绝策略，ThreadPoolExecutor.AbortPolicy
  • 丢弃任务但是不抛异常：ThreadPoolExecutor.DiscardPolicy
  • 丢弃队列最前面的任务，然后重新尝试执行任务，重复此过程：ThreadPoolExecutor.DiscardOldestPolicy
  • 由提交任务的线程来执行任务：ThreadPoolExecutor.CallerRunsPolicy

通过几个实例案例来学习线程池的api

案例1：向线程池中提交任务

使用execute方法、submit方法，向线程池中提交任务

java">public static void main(String[] args) throws ExecutionException, InterruptedException {// 创建线程池，核心线程数2，最大线程数10，存活时间10s，使用数组作为阻塞队列，队列长度是5ThreadPoolExecutor pool = new ThreadPoolExecutor(2, 10, 200, TimeUnit.SECONDS, new ArrayBlockingQueue<>(5));// 提交不需要返回值的任务pool.submit(() -> Utils.println("正在执行 task1"));pool.execute(() -> Utils.println("正在执行 task2"));// 提交需要返回值的任务Future<String> future = pool.submit(() -> "success");System.out.println("future.get() = " + future.get());// 等待所有任务结束后，关闭线程池pool.shutdown();
}

execute和submit方法的区别：

  • execute方法：执行没有返回值的任务。public void execute(Runnable command)
  • submit：可以执行没有返回值的任务，也可以执行有返回值的任务
    • Future<?> submit(Runnable task)
    • <T> Future<T> submit(Callable<T> task)
    • <T> Future<T> submit(Runnable task, T result)
案例2：当阻塞队列是有限阻塞队列时，观察线程池的运行

模拟：一共20个异步任务，每个任务执行耗时1秒，线程池的核心线程池数是2、最大线程数是5、阻塞队列的长度是15。开启一个新的线程，向线程池中提交任务，同时再开启一个线程，打印线程池的状态。根据我们的设置，线程池可以处理完这些任务，观察线程池在处理过程中各个参数的变化。

java">public static void main(String[] args) {// 创建线程池int corePoolSize = 2;int maximumPoolSize = 5;long keepAliveTime = 10;BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(15);ThreadPoolExecutor threadPool = new ThreadPoolExecutor(corePoolSize, maximumPoolSize,keepAliveTime, TimeUnit.SECONDS, workQueue);// 在新线程中向线程池提交任务new Thread(() -> {for (int i = 1; i <= 20; i++) {threadPool.execute(new Runnable() {// 每个任务执行耗时一秒@Overridepublic void run() {try {Thread.sleep(1000L);} catch (InterruptedException e) {throw new RuntimeException(e);}}});}}).start();// 观察线程池的运行状态new Thread(() -> {for (int i = 0; i < 1000; i++) {System.out.println("[" + System.currentTimeMillis() + "]"+ "，核心线程数（参数）" + threadPool.getCorePoolSize()+ "，最大线程数（参数）" + threadPool.getMaximumPoolSize()+ "，当前线程数 = " + threadPool.getPoolSize()+ "，最大线程数（运行时） = " + threadPool.getLargestPoolSize()+ "，正在执行任务的线程数 = " + threadPool.getActiveCount()+ "，已完成任务数 = " + threadPool.getCompletedTaskCount()+ "，任务总数 = " + threadPool.getTaskCount()+ "，缓存队列中的任务数 = " + threadPool.getQueue().size()+ "，线程存活时间 = " + threadPool.getKeepAliveTime(TimeUnit.SECONDS) + "秒");try {Thread.sleep(100);} catch (InterruptedException e) {throw new RuntimeException(e);}}}).start();
}

总结：观察线程池的运行情况，基本符合之前提到的线程池的工作机制。

案例3：当阻塞队列是无界阻塞队列时，观察线程池的运行

模拟：一共20个异步任务，每个任务执行耗时1秒，线程池的核心线程池数是2、最大线程数是5、阻塞队列是无界阻塞队列

java">public static void main(String[] args) {// 创建线程池int corePoolSize = 2;int maximumPoolSize = 5;long keepAliveTime = 10;BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>();ThreadPoolExecutor threadPool = new ThreadPoolExecutor(corePoolSize, maximumPoolSize,keepAliveTime, TimeUnit.SECONDS, workQueue);// 在新线程中向线程池提交任务new Thread(() -> {for (int i = 1; i <= 20; i++) {threadPool.execute(new Runnable() {@Overridepublic void run() {try {Thread.sleep(1000L);} catch (InterruptedException e) {throw new RuntimeException(e);}}});}}).start();// 观察线程池的运行状态new Thread(() -> {for (int i = 0; i < 1000; i++) {System.out.println("[" + System.currentTimeMillis() + "]"+ "，核心线程数（参数）" + threadPool.getCorePoolSize()+ "，最大线程数（参数）" + threadPool.getMaximumPoolSize()+ "，当前线程数 = " + threadPool.getPoolSize()+ "，最大线程数（运行时） = " + threadPool.getLargestPoolSize()+ "，正在执行任务的线程数 = " + threadPool.getActiveCount()+ "，已完成任务数 = " + threadPool.getCompletedTaskCount()+ "，任务总数 = " + threadPool.getTaskCount()+ "，缓存队列中的任务数 = " + threadPool.getQueue().size()+ "，线程存活时间 = " + threadPool.getKeepAliveTime(TimeUnit.SECONDS) + "秒");try {Thread.sleep(100);} catch (InterruptedException e) {throw new RuntimeException(e);}}}).start();}

总结：观察日志中线程池的运行情况，基本可以印证之前提到的线程池的工作机制，无界阻塞队列，只要线程数等于核心线程数了，就不会再创建新线程了。

案例4：关闭线程池shutdown方法
java">public static void main(String[] args) {// 创建线程池int corePoolSize = 2;int maximumPoolSize = 5;long keepAliveTime = 10;BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(15);ThreadPoolExecutor threadPool = new ThreadPoolExecutor(corePoolSize, maximumPoolSize,keepAliveTime, TimeUnit.SECONDS, workQueue);// 在新线程中向线程池提交任务new Thread(() -> {for (int i = 1; i <= 20; i++) {threadPool.execute(new Runnable() {@Overridepublic void run() {for (long i = 0; i < 9000000000L; i++) {}}});}}).start();// 观察线程池的运行状态new Thread(() -> {for (int i = 0; i < 500; i++) {System.out.println("[" + System.currentTimeMillis() + "]"+ "，线程池是否关闭 = " + threadPool.isShutdown()+ "，当前线程数 = " + threadPool.getPoolSize()+ "，正在执行任务的线程数 = " + threadPool.getActiveCount()+ "，已完成任务数 = " + threadPool.getCompletedTaskCount()+ "，任务总数 = " + threadPool.getTaskCount()+ "，缓存队列中的任务数 = " + threadPool.getQueue().size());try {Thread.sleep(100);} catch (InterruptedException e) {throw new RuntimeException(e);}}}).start();try {Thread.sleep(2000);} catch (InterruptedException e) {throw new RuntimeException(e);}System.out.println("关闭线程池");threadPool.shutdown(); // 关闭线程池后，依旧会执行完剩下的任务
}

总结：调用shutdown方法关闭线程池后，依旧会执行完剩下的任务

案例5：关闭线程池 shutdownNow方法
java">public static void main(String[] args) {// 创建线程池int corePoolSize = 2;int maximumPoolSize = 5;long keepAliveTime = 10;BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(15);ThreadPoolExecutor threadPool = new ThreadPoolExecutor(corePoolSize, maximumPoolSize,keepAliveTime, TimeUnit.SECONDS, workQueue);// 在新线程中向线程池提交任务new Thread(() -> {for (int i = 1; i <= 20; i++) {threadPool.execute(new Runnable() {@Overridepublic void run() {for (long i = 0; i < 9000000000L; i++) {}}});}}).start();// 观察线程池的运行状态new Thread(() -> {for (int i = 0; i < 100; i++) {System.out.println("[" + System.currentTimeMillis() + "]"+ "，线程池是否关闭 = " + threadPool.isShutdown()+ "，当前线程数 = " + threadPool.getPoolSize()+ "，最大线程数（运行时） = " + threadPool.getLargestPoolSize()+ "，正在执行任务的线程数 = " + threadPool.getActiveCount()+ "，已完成任务数 = " + threadPool.getCompletedTaskCount()+ "，任务总数 = " + threadPool.getTaskCount()+ "，缓存队列中的任务数 = " + threadPool.getQueue().size()+ "，线程存活时间 = " + threadPool.getKeepAliveTime(TimeUnit.SECONDS) + "秒");try {Thread.sleep(100);} catch (InterruptedException e) {throw new RuntimeException(e);}}}).start();try {Thread.sleep(2000);} catch (InterruptedException e) {throw new RuntimeException(e);}System.out.println("关闭线程池");// 立刻关闭线程池，同时返回阻塞队列中的任务，正在执行任务的线程会被打断，// 如果线程处于休眠状态下，会抛异常，如果线程正在运行，会执行完当前的任务。List<Runnable> runnables = threadPool.shutdownNow();
}

总结：调用shutdownNow方法后，立刻关闭线程池，同时返回阻塞队列中的任务，正在执行任务的线程会被打断，如果线程处于休眠状态下，会抛异常，如果线程正在运行，会执行完当前的任务。

使用经验

如何选择阻塞队列

使用无界阻塞队列要注意内存溢出问题，因为很有可能阻塞了太多任务导致OOM

应该如何选择拒绝策略

当线程池被打满，无法执行新提交的任务时，应该选择何种拒绝策略？可以依据不同的场景，选择不同的拒绝策略。

  • 如果用户需要立刻知道任务失败，选择抛异常，这是默认的拒绝策略
  • 如果用户可以接受延迟，选择使用提交任务的线程来执行任务
  • 非关键性的任务，可以选择丢弃任务并且不抛异常，例如日志系统的收集
  • 如果是实时系统，可以选择丢弃最前面的任务，因为当前的任务价值最高

自定义拒绝策略：可以选择抛异常，并且把任务序列化到磁盘或数据库，系统不忙的时候在启动定时任务执行，尽管会有延迟，但是可以避免丢数据。

线程池的线程数如何设置

首先要看任务的类型是IO密集型，还是CPU密集型？线程数过多，会导致过多的上下文切换；线程数过少，会导致不能充分利用系统资源。

  • IO密集型：频繁读取磁盘上的数据，或者需要通过网络远程调用接口。IO密集型的任务，线程很多时候都会被阻塞，所以多创建一些线程，通常创建2N个线程，其中N代表CPU核数。
  • CPU密集型：非常复杂的调用，循环次数很多，或者递归调用层次很深等。CPU密集型的任务，线程通常不会被阻塞，并且过多的上下文切换会降低效率，所以少创建几个线程，通常创建N + 1个线程，其中N代表CPU核数。

创建线程池的快捷方法 Executors

Executors提供了多个静态方法去创建线程池，在使用上比ThreadPoolExecutor更简单，但是普遍不推荐使用。阿里提供的开发规范中更推荐使用ThreadPoolExecutor来直接创建线程池，这样可以增加对于线程池的控制力度。

Executors提供的创建线程池的方法：

  • 固定数目的线程池：newFixedThreadPool，创建一个线程池，核心线程数和最大线程数的值是相等的，使用无界阻塞队列
java">// 用户指定线程数即可，
public static ExecutorService newFixedThreadPool(int nThreads) {return new ThreadPoolExecutor(nThreads, nThreads,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>());
}
  • 可缓存线程池：newCachedThreadPool，将核心线程数设置为0，将最大线程数设置为int类型的最大值，来了任务就创建线程运行，当线程空闲超过60秒，就销毁线程，阻塞队列中不存储元素，来了任务，就创建新线程执行或使用原有的空闲线程执行。
java">public static ExecutorService newCachedThreadPool() {return new ThreadPoolExecutor(0, Integer.MAX_VALUE,60L, TimeUnit.SECONDS,new SynchronousQueue<Runnable>());
}
  • 单个线程的线程池：newSingleThreadExecutor，线程池中只有一个线程，使用无界阻塞队列
java">public static ExecutorService newSingleThreadExecutor() {return new FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>()));
}

定时任务线程池 ScheduledThreadPoolExecutor

ScheduledThreadPoolExecutor，可以执行定时任务的线程池，它继承了ThreadPoolExecutor，添加了定时任务的功能。

通过几个实例案例来学习定时任务的api

案例1：只执行一次的定时任务

schedule方法：public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit)：通过schedule方法提交的定时任务，只会在指定时间后执行一次

java">public static void main(String[] args) {int corePoolSize = 3;ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(corePoolSize);// 执行定时任务，在指定时间后执行一次，这里是在任务提交后的3秒后执行一次打印任务System.out.println(System.currentTimeMillis() + " 开始计时");executor.schedule(() -> {System.out.println(System.currentTimeMillis() + " hello world");}, 3L, TimeUnit.SECONDS);executor.shutdown();
}
案例2：以固定频率重复执行的定时任务

scheduleAtFixedRate：public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit)：定时任务会以固定的频率执行

java">public static void main(String[] args) {int corePoolSize = 3;ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(corePoolSize);Utils.println("任务开始");// 任务提交后延迟1秒，并且每秒执行1次AtomicInteger i = new AtomicInteger(1);executor.scheduleAtFixedRate(() -> {int taskId = i.get();Utils.println("任务" + taskId + "开始"); // 工具方法，打印字符串的时候同时打印当前时间、线程try {Thread.sleep(new Random().nextInt(5) * 1000L);} catch (InterruptedException e) {throw new RuntimeException(e);}Utils.println("任务" + taskId + "结束");i.set(taskId + 1);}, 1L, 1L, TimeUnit.SECONDS);
}

注意：如果前一个定时任务在下一个任务应该开始执行的时候还没有执行完成，下一个定时任务的执行时间会被延迟，并且它会在前一个定时任务执行完成后立刻执行

案例3：以固定延迟重复执行的定时任务

scheduleWithFixedDelay：public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit)：和scheduleAtFixedRate方法的功能类似，但是任务的下次执行时间受上次任务的执行时间的影响，因为它是上次任务执行完成之后，间隔指定时长，再执行下次任务

java">public static void main(String[] args) {int corePoolSize = 3;ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(corePoolSize);AtomicInteger i = new AtomicInteger(1);executor.scheduleWithFixedDelay(() -> {int taskId = i.get();Utils.println("任务" + taskId + "开始");try {Thread.sleep(new Random().nextInt(5) * 1000L);} catch (InterruptedException e) {throw new RuntimeException(e);}Utils.println("任务" + taskId + "结束");i.set(taskId + 1);}, 1000L, 1000L, TimeUnit.MILLISECONDS);
}
实战案例：写一个每晚0点定时执行的定时任务

方法：先计算出当前时间到下一天0的之间的秒数，作为启动时的延迟，然后使用1天的秒数作为固定频率。

java">// 每天晚上0点固定执行的定时任务
public static void main(String[] args) {int corePoolSize = 3;ScheduledThreadPoolExecutor poolExecutor = new ScheduledThreadPoolExecutor(corePoolSize);poolExecutor.scheduleAtFixedRate(() -> {Utils.println("任务");}, getSecondsToNextDayMidnight(), 86400L, TimeUnit.SECONDS);
}// 获取当前时间到第二天0点的秒数
public static long getSecondsToNextDayMidnight() {// 获取当前时间LocalDateTime now = LocalDateTime.now();// 获取下一天0点的时间LocalDateTime nextDayMidnight = now.plusDays(1).with(LocalTime.MIDNIGHT);// 计算两个时间之间的秒数return ChronoUnit.SECONDS.between(now, nextDayMidnight);
}
案例5：有返回值的定时任务

定时任务使用ScheduledFuture来接收返回值，它继承了Future接口

java">public interface ScheduledFuture<V> extends Delayed, Future<V> { }

案例：

java">// 有返回值的定时任务
public static void main(String[] args) {int corePoolSize = 3;ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(corePoolSize);// 执行定时任务，在指定时间后执行一次System.out.println(System.currentTimeMillis() + " 开始计时");ScheduledFuture<String> scheduledFuture = executor.schedule(new Callable<String>() {@Overridepublic String call() throws Exception {System.out.println(System.currentTimeMillis() + " hello world");return "result";}}, 3L, TimeUnit.SECONDS);try {String s = scheduledFuture.get();System.out.println("定时任务的结果 = " + s);} catch (InterruptedException | ExecutionException e) {throw new RuntimeException(e);}executor.shutdown();
}

Q&A

1、核心线程数以外的线程，是怎么实现超过存活时间就移除的？

线程不执行任务时会一直阻塞地获取异步任务，如果当前线程数大于核心线程池数，线程在获取异步任务时会加上阻塞时长，超过阻塞时长，线程就会唤醒，此时会判断，如果线程没有获取到异步任务，就会从线程池中移出线程，这个阻塞时长就是线程的存活时长。


http://www.ppmy.cn/server/172806.html

相关文章

JAVA入门——反射

JAVA入门——反射

一、什么是反射 反射允许对封装类的字段、方法和构造函数的信息进行编程访问我们是从class字节码文件中获取这些信息的&#xff0c;所以先学习如何获取class文件 二、获取class对象的三种方式 Class.forName("全类名") 类名.class 对象.getClass() 例&#xff1a; …
阅读更多...
【LeetCode226】翻转二叉树

【LeetCode226】翻转二叉树

题目描述 给你一棵二叉树的根节点 root &#xff0c;翻转这棵二叉树&#xff0c;并返回其根节点。 思路与算法 这个问题自然是递归的&#xff0c;因为反转一棵树涉及到反转它的子树。 让 f(node) 是一个函数&#xff0c;用于反转以 node 为根的二叉树。如果 node 有左子树 L…
阅读更多...
【Spring AOP】_切点类的切点表达式

【Spring AOP】_切点类的切点表达式

目录 1. 根据方法签名匹配编写切点表达式 1.1 具体语法 1.2 通配符表达规范 2. 根据注解匹配编写切点表达式 2.1 实现步骤 2.2 元注解及其常用取值含义 2.3 使用自定义注解 2.3.1 编写自定义注解MyAspect 2.3.2 编写切面类MyAspectDemo 2.3.3 编写测试类及测试方法 在…
阅读更多...
Python的枚举enumerate学习

Python的枚举enumerate学习

这个enumerate可以说是非常常用和强大的一个东西&#xff0c;值得单独盘点学习复习一下。 关键语法拆解 for i, (doc, meta, doc_id) in enumerate(zip(docs, metas, ids)):# 处理每个元素的代码1️⃣ zip(docs, metas, ids) 功能&#xff1a;将3个列表逐元素打包成元组示例…
阅读更多...
LC串联带初始值的时域表达式

LC串联带初始值的时域表达式

LC串联&#xff0c;在t0时刻接入直流电压 U i n U_{in} Uin​。 电感电流 i ( t ) i(t) i(t)和电容电压 u c ( t ) u_c(t) uc​(t)的时域表达式可通过二阶微分方程求解。以下是推导过程与结果&#xff1a; 1. 微分方程建立 电感 L L L与电容 C C C串联&#xff0c;接入直流…
阅读更多...
如何为 Power Automate 配置 Azure Key Vault 权限

如何为 Power Automate 配置 Azure Key Vault 权限

前言 最近&#xff0c;在Power Automate中使用Azure Key Vault&#xff0c;然后&#xff0c;就需要配置一下AKV的权限。 正文 1.我们在Azure Portal里新建一个Key vault&#xff0c;如下图&#xff1a; 2.进入Access policies&#xff0c;点击Create&#xff0c;如下图&#xf…
阅读更多...
网上花店微信小程序+论文源码调试讲解

网上花店微信小程序+论文源码调试讲解

第四章 系统设计 4.1 总体功能 网上花店微信小程序是根据需求定制开发&#xff0c;开发软件选用IDEA平台配合MySQL数据库进行开发环境的搭建操作&#xff0c;网站采用WEB应用程序中最流行的小程序结构进行开发&#xff0c;用户访问系统数据仅仅需要在客户端安装谷歌浏览器或者…
阅读更多...
Hi3516CV610电瓶车检测 电动自行车检测 人脸检测 人形检测 车辆检测 宠物检测 包裹检测 源码

Hi3516CV610电瓶车检测 电动自行车检测 人脸检测 人形检测 车辆检测 宠物检测 包裹检测 源码

海思全线一代AI摄像机SoC&#xff0c;在板端实现实时AI检测&#xff0c;视频传输、视频录像、报警等功能 Hi3516CV610开发板实时AI检测效果演示&#xff0c;实时在板端完成 视频采集—AI检测—视频编码—网络传输&#xff0c;AI检测算法无需license无需授权&#xff0c;终身免…
阅读更多...
最新文章

Copyright @ 2022~2023