首页 > 编程语言 >并发编程:线程池(下)

并发编程:线程池(下)

时间:2024-09-10 10:52:50浏览次数:9  
标签:队列 核心 编程 并发 任务 线程 切换 CPU

一、线程池常用的阻塞队列有哪些?

新任务来的时候会先判断当前运行的线程数量是否达到核心线程数,如果达到的话,新任务就会被存放在队列中。

不同的线程池会选用不同的阻塞队列,我们可以结合内置线程池来分析。

  • 容量为 Integer.MAX_VALUELinkedBlockingQueue(有界阻塞队列):FixedThreadPoolSingleThreadExecutorFixedThreadPool最多只能创建核心线程数的线程(核心线程数和最大线程数相等),SingleThreadExecutor只能创建一个线程(核心线程数和最大线程数都是 1),二者的任务队列永远不会被放满。
  • SynchronousQueue(同步队列):CachedThreadPoolSynchronousQueue 没有容量,不存储元素,目的是保证对于提交的任务,如果有空闲线程,则使用空闲线程来处理;否则新建一个线程来处理任务。也就是说,CachedThreadPool 的最大线程数是 Integer.MAX_VALUE ,可以理解为线程数是可以无限扩展的,可能会创建大量线程,从而导致 OOM。
  • DelayedWorkQueue(延迟队列):ScheduledThreadPoolSingleThreadScheduledExecutorDelayedWorkQueue 的内部元素并不是按照放入的时间排序,而是会按照延迟的时间长短对任务进行排序,内部采用的是“堆”的数据结构,可以保证每次出队的任务都是当前队列中执行时间最靠前的。DelayedWorkQueue 添加元素满了之后会自动扩容,增加原来容量的 50%,即永远不会阻塞,最大扩容可达 Integer.MAX_VALUE,所以最多只能创建核心线程数的线程。
  • ArrayBlockingQueue(有界阻塞队列):底层由数组实现,容量一旦创建,就不能修改。

二、线程池处理任务的流程

  1. 如果当前运行的线程数小于核心线程数,那么就会新建一个线程来执行任务。
  2. 如果当前运行的线程数等于或大于核心线程数,但是小于最大线程数,那么就把该任务放入到任务队列里等待执行。
  3. 如果向任务队列投放任务失败(任务队列已经满了),但是当前运行的线程数是小于最大线程数的,就新建一个线程来执行任务。
  4. 如果当前运行的线程数已经等同于最大线程数了,新建线程将会使当前运行的线程超出最大线程数,那么当前任务会被拒绝,拒绝策略会调用RejectedExecutionHandler.rejectedExecution()方法。

再提一个有意思的小问题:线程池在提交任务前,可以提前创建线程吗?

答案是可以的!ThreadPoolExecutor 提供了两个方法帮助我们在提交任务之前,完成核心线程的创建,从而实现线程池预热的效果:

  • prestartCoreThread():启动一个线程,等待任务,如果已达到核心线程数,这个方法返回 false,否则返回 true;
  • prestartAllCoreThreads():启动所有的核心线程,并返回启动成功的核心线程数。

三、线程池中线程异常后,销毁还是复用?

直接说结论,需要分两种情况:

  • 使用execute()提交任务:当任务通过execute()提交到线程池并在执行过程中抛出异常时,如果这个异常没有在任务内被捕获,那么该异常会导致当前线程终止,并且异常会被打印到控制台或日志文件中。线程池会检测到这种线程终止,并创建一个新线程来替换它,从而保持配置的线程数不变。
  • 使用submit()提交任务:对于通过submit()提交的任务,如果在任务执行中发生异常,这个异常不会直接打印出来。相反,异常会被封装在由submit()返回的Future对象中。当调用Future.get()方法时,可以捕获到一个ExecutionException。在这种情况下,线程不会因为异常而终止,它会继续存在于线程池中,准备执行后续的任务。

简单来说:使用execute()时,未捕获异常导致线程终止,线程池创建新线程替代;使用submit()时,异常被封装在Future中,线程继续复用。

这种设计允许submit()提供更灵活的错误处理机制,因为它允许调用者决定如何处理异常,而execute()则适用于那些不需要关注执行结果的场景。

四、如何给线程池命名?

初始化线程池的时候需要显示命名(设置线程池名称前缀),有利于定位问题。

默认情况下创建的线程名字类似 pool-1-thread-n 这样的,没有业务含义,不利于我们定位问题。

给线程池里的线程命名通常有下面两种方式:

1.利用 guava 的 ThreadFactoryBuilder

ThreadFactory threadFactory = new ThreadFactoryBuilder()
                        .setNameFormat(threadNamePrefix + "-%d")
                        .setDaemon(true).build();
ExecutorService threadPool = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, TimeUnit.MINUTES, workQueue, threadFactory);

2.自己实现 ThreadFactory

import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 线程工厂,它设置线程名称,有利于我们定位问题。
 */
public final class NamingThreadFactory implements ThreadFactory {

    private final AtomicInteger threadNum = new AtomicInteger();
    private final String name;

    /**
     * 创建一个带名字的线程池生产工厂
     */
    public NamingThreadFactory(String name) {
        this.name = name;
    }

    @Override
    public Thread newThread(Runnable r) {
        Thread t = new Thread(r);
        t.setName(name + " [#" + threadNum.incrementAndGet() + "]");
        return t;
    }
}

五、如何设定线程池的大小?

很多人甚至可能都会觉得把线程池配置过大一点比较好!我觉得这明显是有问题的。就拿我们生活中非常常见的一例子来说:并不是人多就能把事情做好,增加了沟通交流成本。你本来一件事情只需要 3 个人做,你硬是拉来了 6 个人,会提升做事效率嘛?我想并不会。 线程数量过多的影响也是和我们分配多少人做事情一样,对于多线程这个场景来说主要是增加了上下文切换成本。不清楚什么是上下文切换的话,可以看我下面的介绍。

上下文切换:

多线程编程中一般线程的个数都大于 CPU 核心的个数,而一个 CPU 核心在任意时刻只能被一个线程使用,为了让这些线程都能得到有效执行,CPU 采取的策略是为每个线程分配时间片并轮转的形式。当一个线程的时间片用完的时候就会重新处于就绪状态让给其他线程使用,这个过程就属于一次上下文切换。概括来说就是:当前任务在执行完 CPU 时间片切换到另一个任务之前会先保存自己的状态,以便下次再切换回这个任务时,可以再加载这个任务的状态。任务从保存到再加载的过程就是一次上下文切换

上下文切换通常是计算密集型的。也就是说,它需要相当可观的处理器时间,在每秒几十上百次的切换中,每次切换都需要纳秒量级的时间。所以,上下文切换对系统来说意味着消耗大量的 CPU 时间,事实上,可能是操作系统中时间消耗最大的操作。

Linux 相比与其他操作系统(包括其他类 Unix 系统)有很多的优点,其中有一项就是,其上下文切换和模式切换的时间消耗非常少。

类比于实现世界中的人类通过合作做某件事情,我们可以肯定的一点是线程池大小设置过大或者过小都会有问题,合适的才是最好。

  • 如果我们设置的线程池数量太小的话,如果同一时间有大量任务/请求需要处理,可能会导致大量的请求/任务在任务队列中排队等待执行,甚至会出现任务队列满了之后任务/请求无法处理的情况,或者大量任务堆积在任务队列导致 OOM。这样很明显是有问题的,CPU 根本没有得到充分利用。
  • 如果我们设置线程数量太大,大量线程可能会同时在争取 CPU 资源,这样会导致大量的上下文切换,从而增加线程的执行时间,影响了整体执行效率。

有一个简单并且适用面比较广的公式:

  • CPU 密集型任务(N+1): 这种任务消耗的主要是 CPU 资源,可以将线程数设置为 N(CPU 核心数)+1。比 CPU 核心数多出来的一个线程是为了防止线程偶发的缺页中断,或者其它原因导致的任务暂停而带来的影响。一旦任务暂停,CPU 就会处于空闲状态,而在这种情况下多出来的一个线程就可以充分利用 CPU 的空闲时间。
  • I/O 密集型任务(2N): 这种任务应用起来,系统会用大部分的时间来处理 I/O 交互,而线程在处理 I/O 的时间段内不会占用 CPU 来处理,这时就可以将 CPU 交出给其它线程使用。因此在 I/O 密集型任务的应用中,我们可以多配置一些线程,具体的计算方法是 2N。

如何判断是 CPU 密集任务还是 IO 密集任务?

CPU 密集型简单理解就是利用 CPU 计算能力的任务比如你在内存中对大量数据进行排序。但凡涉及到网络读取,文件读取这类都是 IO 密集型,这类任务的特点是 CPU 计算耗费时间相比于等待 IO 操作完成的时间来说很少,大部分时间都花在了等待 IO 操作完成上。

线程数更严谨的计算的方法应该是:最佳线程数 = N(CPU 核心数)∗(1+WT(线程等待时间)/ST(线程计算时间)),其中 WT(线程等待时间)=线程运行总时间 - ST(线程计算时间)

线程等待时间所占比例越高,需要越多线程。线程计算时间所占比例越高,需要越少线程。

我们可以通过 JDK 自带的工具 VisualVM 来查看 WT/ST 比例。

CPU 密集型任务的 WT/ST 接近或者等于 0,因此, 线程数可以设置为 N(CPU 核心数)∗(1+0)= N,和我们上面说的 N(CPU 核心数)+1 差不多。

IO 密集型任务下,几乎全是线程等待时间,从理论上来说,你就可以将线程数设置为 2N(按道理来说,WT/ST 的结果应该比较大,这里选择 2N 的原因应该是为了避免创建过多线程吧)。

 公式也只是参考,具体还是要根据项目实际线上运行情况来动态调整。

六、如何动态修改线程池的参数?

美团技术团队在《Java 线程池实现原理及其在美团业务中的实践》open in new window这篇文章中介绍到对线程池参数实现可自定义配置的思路和方法。

美团技术团队的思路是主要对线程池的核心参数实现自定义可配置。这三个核心参数是:

  • corePoolSize : 核心线程数线程数定义了最小可以同时运行的线程数量。
  • maximumPoolSize : 当队列中存放的任务达到队列容量的时候,当前可以同时运行的线程数量变为最大线程数。
  • workQueue: 当新任务来的时候会先判断当前运行的线程数量是否达到核心线程数,如果达到的话,新任务就会被存放在队列中。

为什么是这三个参数?

这三个参数是 ThreadPoolExecutor 最重要的参数,它们基本决定了线程池对于任务的处理策略。

格外需要注意的是corePoolSize, 程序运行期间的时候,我们调用 setCorePoolSize()这个方法的话,线程池会首先判断当前工作线程数是否大于corePoolSize,如果大于的话就会回收工作线程。

另外,你也看到了上面并没有动态指定队列长度的方法,美团的方式是自定义了一个叫做 ResizableCapacityLinkedBlockIngQueue 的队列(主要就是把LinkedBlockingQueue的 capacity 字段的 final 关键字修饰给去掉了,让它变为可变的)。

七、如何设计一个能够根据任务的优先级来执行的线程池?

这是一个常见的面试问题,本质其实还是在考察求职者对于线程池以及阻塞队列的掌握。

我们上面也提到了,不同的线程池会选用不同的阻塞队列作为任务队列,比如FixedThreadPool 使用的是LinkedBlockingQueue(有界队列),默认构造器初始的队列长度为 Integer.MAX_VALUE ,由于队列永远不会被放满,因此FixedThreadPool最多只能创建核心线程数的线程。

假如我们需要实现一个优先级任务线程池的话,那可以考虑使用 PriorityBlockingQueue (优先级阻塞队列)作为任务队列(ThreadPoolExecutor 的构造函数有一个 workQueue 参数可以传入任务队列)。

PriorityBlockingQueue 是一个支持优先级的无界阻塞队列,可以看作是线程安全的 PriorityQueue,两者底层都是使用小顶堆形式的二叉堆,即值最小的元素优先出队。不过,PriorityQueue 不支持阻塞操作。

要想让 PriorityBlockingQueue 实现对任务的排序,传入其中的任务必须是具备排序能力的,方式有两种:

  1. 提交到线程池的任务实现 Comparable 接口,并重写 compareTo 方法来指定任务之间的优先级比较规则。
  2. 创建 PriorityBlockingQueue 时传入一个 Comparator 对象来指定任务之间的排序规则(推荐)。

不过,这存在一些风险和问题,比如:

  • PriorityBlockingQueue 是无界的,可能堆积大量的请求,从而导致 OOM。
  • 可能会导致饥饿问题,即低优先级的任务长时间得不到执行。
  • 由于需要对队列中的元素进行排序操作以及保证线程安全(并发控制采用的是可重入锁 ReentrantLock),因此会降低性能。

对于 OOM 这个问题的解决比较简单粗暴,就是继承PriorityBlockingQueue 并重写一下 offer 方法(入队)的逻辑,当插入的元素数量超过指定值就返回 false 。

饥饿问题这个可以通过优化设计来解决(比较麻烦),比如等待时间过长的任务会被移除并重新添加到队列中,但是优先级会被提升。

对于性能方面的影响,是没办法避免的,毕竟需要对任务进行排序操作。并且,对于大部分业务场景来说,这点性能影响是可以接受的。

标签:队列,核心,编程,并发,任务,线程,切换,CPU
From: https://blog.csdn.net/AWSDN/article/details/142091923

相关文章

  • 多线程模拟叫号看病
    //普通号publicclassNormalThreadextendsThread{privateintnum=20;publicintgetNum(){returnnum;}publicvoidsetNum(intnum){this.num=num;}publicNormalThread(Stringname,intnum){super(......
  • 并发编程:ThreadLocal
    一、ThreadLocal有什么用?通常情况下,我们创建的变量是可以被任何一个线程访问并修改的。如果想实现每一个线程都有自己的专属本地变量该如何解决呢?JDK中自带的ThreadLocal类正是为了解决这样的问题。ThreadLocal类主要解决的就是让每个线程绑定自己的值,可以将ThreadLocal类......
  • Java并发编程实战 11 | 线程活跃问题(死锁,活锁和饥饿)
    并发应用程序的“活跃度”指的是它及时执行并完成任务的能力。活跃性问题则是指程序无法最终得到预期的运行结果。相比于线程安全问题,存活性问题可能会导致更严重的后果。例如,死锁会使程序完全停滞,导致无法继续执行。常见的活跃性问题包括以下三种:1.死锁(Deadlock)死锁发......
  • 一个例子彻底搞懂对线程模型的理解 !
    如何理解一对一线程模型假设你的计算机是一个大工厂,操作系统(如Windows或Linux)是工厂的管理系统,而Java虚拟机(JVM)是工厂里的一个部门。这些部门里有很多工人(线程),每个工人可以负责一项具体的任务(比如处理一个计算或者操作某个数据)。在Java中,每当你创建一个线程时,JVM就会告诉......
  • 几种并发模式的实现
    常见的几个需要并发的场景,具体的实现方式asyncio异步并发importasyncioasyncdeftask(id,delay):awaitasyncio.sleep(delay)returnf"Task{id}completed"asyncdefmain():tasks=[task(1,2),task(2,1),task(3,3)]done,pending=await......
  • Spring Cloud全解析:熔断之Hystrix线程隔离导致的问题
    Hystrix线程隔离在微服务框架中,可能一个服务需要调用多个微服务,在tomcat中运行时,tomcat只是分配了100个线程,由于多个服务之间调用的时间消耗过长,可能会导致线程耗尽,而在Hystrix中存在线程隔离,对于每个微服务分配一个线程池,访问某个微服务时就从对应的线程池中取线程,如果对应线程......
  • day04(网络编程基础)tcp编程
    目录tcp编程流程服务器客户端函数接口socketbindlistenaccept​​​​​​​recv​​​​​​​connect​​​​​​​send初始版服务器客户端 加功能:1.客户端连接成功后进入循环发送状态,从终端获取用户输入并发送,当用户输入“quit”字符后退出循环并关闭客......
  • 使用 Parallel 类进行多线程编码(下)
    2.Parallel.ForEach()的使用 从ForEach()这个名字可以看出该方法是用来遍历泛型集合的,新建一个ASP.NETCore Web应用的项目,如下:         在Index.cshtml.cs文件中增加一个UserInfo.cs的类,代码如下:publicclassUserInfo{publicint......
  • 全网最适合入门的面向对象编程教程:46 Python函数方法与接口-函数与事件驱动框架
    全网最适合入门的面向对象编程教程:46Python函数方法与接口-函数与事件驱动框架摘要:函数是Python中的一等公民,是一种可重用的代码块,用于封装特定的逻辑;事件驱动框架是一种编程模式,它将程序的控制流转移给外部事件,如用户输入、系统消息等,它事件驱动框架可以使用函数作为事......
  • Numba最近邻插值(CPU+ GPU + Z轴切块 + XYZ轴切块 + 多线程)
    文章目录最近邻插值(加速方法)(1)scipy.ndimage.zoom(2)Numba-CPU加速(3)Numba-GPU加速(4)Numba-CPU加速(Z轴切块)(5)Numba-CPU加速(XYZ轴切块)(6)Numba-CPU加速(XYZ轴切块)+多线程输入数据插值倍数时耗scipy.ndimage.zoom(1024,1024,512)4172.16sNumba-CPU(1024,1024,512)456.58sN......