Loading [MathJax]/jax/output/CommonHTML/config.js
前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
圈层
工具
发布
首页
学习
活动
专区
圈层
工具
MCP广场
社区首页 >专栏 >Executors源码之线程池

Executors源码之线程池

作者头像
二十三年蝉
发布于 2021-06-29 03:10:06
发布于 2021-06-29 03:10:06
44700
代码可运行
举报
文章被收录于专栏:闻道于事闻道于事
运行总次数:0
代码可运行

首先,看下四种线程池的八种构造方法:

代码语言:javascript
代码运行次数:0
运行
AI代码解释
复制
/**
     * Creates a thread pool that reuses a fixed number of threads
     * operating off a shared unbounded queue.  At any point, at most
     * {@code nThreads} threads will be active processing tasks.
     * If additional tasks are submitted when all threads are active,
     * they will wait in the queue until a thread is available.
     * If any thread terminates due to a failure during execution
     * prior to shutdown, a new one will take its place if needed to
     * execute subsequent tasks.  The threads in the pool will exist
     * until it is explicitly {@link ExecutorService#shutdown shutdown}.
     *
     * @param nThreads the number of threads in the pool
     * @return the newly created thread pool
     * @throws IllegalArgumentException if {@code nThreads <= 0}
     */
    public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }

    /**
     * Creates a thread pool that maintains enough threads to support
     * the given parallelism level, and may use multiple queues to
     * reduce contention. The parallelism level corresponds to the
     * maximum number of threads actively engaged in, or available to
     * engage in, task processing. The actual number of threads may
     * grow and shrink dynamically. A work-stealing pool makes no
     * guarantees about the order in which submitted tasks are
     * executed.
     *
     * @param parallelism the targeted parallelism level
     * @return the newly created thread pool
     * @throws IllegalArgumentException if {@code parallelism <= 0}
     * @since 1.8
     */
    public static ExecutorService newWorkStealingPool(int parallelism) {
        return new ForkJoinPool
            (parallelism,
             ForkJoinPool.defaultForkJoinWorkerThreadFactory,
             null, true);
    }

    /**
     * Creates a work-stealing thread pool using all
     * {@link Runtime#availableProcessors available processors}
     * as its target parallelism level.
     * @return the newly created thread pool
     * @see #newWorkStealingPool(int)
     * @since 1.8
     */
    public static ExecutorService newWorkStealingPool() {
        return new ForkJoinPool
            (Runtime.getRuntime().availableProcessors(),
             ForkJoinPool.defaultForkJoinWorkerThreadFactory,
             null, true);
    }

    /**
     * Creates a thread pool that reuses a fixed number of threads
     * operating off a shared unbounded queue, using the provided
     * ThreadFactory to create new threads when needed.  At any point,
     * at most {@code nThreads} threads will be active processing
     * tasks.  If additional tasks are submitted when all threads are
     * active, they will wait in the queue until a thread is
     * available.  If any thread terminates due to a failure during
     * execution prior to shutdown, a new one will take its place if
     * needed to execute subsequent tasks.  The threads in the pool will
     * exist until it is explicitly {@link ExecutorService#shutdown
     * shutdown}.
     *
     * @param nThreads the number of threads in the pool
     * @param threadFactory the factory to use when creating new threads
     * @return the newly created thread pool
     * @throws NullPointerException if threadFactory is null
     * @throws IllegalArgumentException if {@code nThreads <= 0}
     */
    public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>(),
                                      threadFactory);
    }

    /**
     * Creates an Executor that uses a single worker thread operating
     * off an unbounded queue. (Note however that if this single
     * thread terminates due to a failure during execution prior to
     * shutdown, a new one will take its place if needed to execute
     * subsequent tasks.)  Tasks are guaranteed to execute
     * sequentially, and no more than one task will be active at any
     * given time. Unlike the otherwise equivalent
     * {@code newFixedThreadPool(1)} the returned executor is
     * guaranteed not to be reconfigurable to use additional threads.
     *
     * @return the newly created single-threaded Executor
     */
    public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }

    /**
     * Creates an Executor that uses a single worker thread operating
     * off an unbounded queue, and uses the provided ThreadFactory to
     * create a new thread when needed. Unlike the otherwise
     * equivalent {@code newFixedThreadPool(1, threadFactory)} the
     * returned executor is guaranteed not to be reconfigurable to use
     * additional threads.
     *
     * @param threadFactory the factory to use when creating new
     * threads
     *
     * @return the newly created single-threaded Executor
     * @throws NullPointerException if threadFactory is null
     */
    public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>(),
                                    threadFactory));
    }

    /**
     * Creates a thread pool that creates new threads as needed, but
     * will reuse previously constructed threads when they are
     * available.  These pools will typically improve the performance
     * of programs that execute many short-lived asynchronous tasks.
     * Calls to {@code execute} will reuse previously constructed
     * threads if available. If no existing thread is available, a new
     * thread will be created and added to the pool. Threads that have
     * not been used for sixty seconds are terminated and removed from
     * the cache. Thus, a pool that remains idle for long enough will
     * not consume any resources. Note that pools with similar
     * properties but different details (for example, timeout parameters)
     * may be created using {@link ThreadPoolExecutor} constructors.
     *
     * @return the newly created thread pool
     */
    public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }

    /**
     * Creates a thread pool that creates new threads as needed, but
     * will reuse previously constructed threads when they are
     * available, and uses the provided
     * ThreadFactory to create new threads when needed.
     * @param threadFactory the factory to use when creating new threads
     * @return the newly created thread pool
     * @throws NullPointerException if threadFactory is null
     */
    public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>(),
                                      threadFactory);
    }

newSingleThreadExecutor 创建一个单线程化的线程池,它只会用唯一的工作线程来执行任务,保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。

newFixedThreadPool 创建一个定长线程池,可控制线程最大并发数,超出的线程会在队列中等待。

newScheduledThreadPool 创建一个可定期或者延时执行任务的定长线程池,支持定时及周期性任务执行。

newCachedThreadPool 创建一个可缓存线程池,如果线程池长度超过处理需要,可灵活回收空闲线程,若无可回收,则新建线程。

四种线程池都是创建了ThreadPollExecutor对象,只是传递的参数不一样而已,观察传入的workQueue 都是默认,即最大可添加Integer.MAX_VALUE个任务,这也就是阿里巴巴java开发规范禁止直接使用java提供的默认线程池的原因了

ThreadPoolExecutor构造参数

int corePoolSize, 核心线程大小 int maximumPoolSize,最大线程大小 long keepAliveTime, 超过corePoolSize的线程多久不活动被销毁时间 TimeUnit unit,时间单位 BlockingQueue<Runnable> workQueue 任务队列 ThreadFactory threadFactory 线程池工厂 RejectedExecutionHandler handler 拒绝策略

其中任务队列:

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

线程池I的执行流程

当线程池小于corePoolSize时,新提交任务将创建一个新线程执行任务,即使此时线程池中存在空闲线程。 当线程池达到corePoolSize时,新提交任务将被放入workQueue中,等待线程池中任务调度执行 当workQueue已满,且maximumPoolSize>corePoolSize时,新提交任务会创建新线程执行任务 当提交任务数超过maximumPoolSize时,新提交任务由RejectedExecutionHandler处理 当线程池中超过corePoolSize线程,空闲时间达到keepAliveTime时,释放空闲线程 当设置allowCoreThreadTimeOut(true)时,该参数默认false,线程池中corePoolSize线程空闲时间达到keepAliveTime也将关闭

不同的使用场景

newFixedThreadPool:

  • 底层:返回ThreadPoolExecutor实例,接收参数为所设定线程数量n,corePoolSize和maximumPoolSize均为n;keepAliveTime为0L;时间单位TimeUnit.MILLISECONDS;WorkQueue为:new LinkedBlockingQueue<Runnable>() 无界阻塞队列
  • 通俗:创建可容纳固定数量线程的池子,每个线程的存活时间是无限的,当池子满了就不再添加线程了;如果池中的所有线程均在繁忙状态,对于新任务会进入阻塞队列中(无界的阻塞队列)
  • 适用:执行长期任务

newSingleThreadExecutor:

  • 底层:FinalizableDelegatedExecutorService包装的ThreadPoolExecutor实例,corePoolSize为1;maximumPoolSize为1;keepAliveTime为0L;时间单位TimeUnit.MILLISECONDS;workQueue为:new LinkedBlockingQueue<Runnable>() 无解阻塞队列
  • 通俗:创建只有一个线程的线程池,当该线程正繁忙时,对于新任务会进入阻塞队列中(无界的阻塞队列)
  • 适用:按顺序执行任务的场景

newCachedThreadPool:

  • 底层:返回ThreadPoolExecutor实例,corePoolSize为0;maximumPoolSize为Integer.MAX_VALUE;keepAliveTime为60L;时间单位TimeUnit.SECONDS;workQueue为SynchronousQueue(同步队列)
  • 通俗:当有新任务到来,则插入到SynchronousQueue中,由于SynchronousQueue是同步队列,因此会在池中寻找可用线程来执行,若有可以线程则执行,若没有可用线程则创建一个线程来执行该任务;若池中线程空闲时间超过指定时间,则该线程会被销毁。
  • 适用:执行很多短期的异步任务

NewScheduledThreadPool:

  • 底层:创建ScheduledThreadPoolExecutor实例,该对象继承了ThreadPoolExecutor,corePoolSize为传递来的参数,maximumPoolSize为Integer.MAX_VALUE;keepAliveTime为0;时间单位TimeUnit.NANOSECONDS;workQueue为:new DelayedWorkQueue() 一个按超时时间升序排序的队列
  • 通俗:创建一个固定大小的线程池,线程池内线程存活时间无限制,线程池可以支持定时及周期性任务执行,如果所有线程均处于繁忙状态,对于新任务会进入DelayedWorkQueue队列中,这是一种按照超时时间排序的队列结构
  • 适用:执行周期性任务

注:

  • 一般如果线程池任务队列采用LinkedBlockingQueue队列的话,那么不会拒绝任何任务(因为其大小为Integer.MAX_VALUE),这种情况下,ThreadPoolExecutor最多仅会按照最小线程数corePoolSize来创建线程,也就是说线程池大小被忽略了。
  • 如果线程池任务队列采用ArrayBlockingQueue队列,初始化设置了最大队列数。那么ThreadPoolExecutor的maximumPoolSize才会生效,那么ThreadPoolExecutor的maximumPoolSize才会生效会采用新的算法处理任务,
  • 例如假定线程池的最小线程数为4,最大为8,ArrayBlockingQueue最大为10。随着任务到达并被放到队列中,线程池中最多运行4个线程(即核心线程数)直到队列完全填满,也就是说等待状态的任务小于等于10,ThreadPoolExecutor也只会利用4个核心线程线程处理任务。
  • 如果队列已满,而又有新任务进来,此时才会启动一个新线程,这里不会因为队列已满而拒接该任务,相反会启动一个新线程。新线程会运行队列中的第一个任务,为新来的任务腾出空间。如果线程数已经等于最大线程数,任务队列也已经满了,则线程池会拒绝这个任务,默认拒绝策略是抛出异常。
  • 这个算法背的理念是:该池大部分时间仅使用核心线程(4个),即使有适量的任务在队列中等待运行。这时线程池就可以用作节流阀。如果挤压的请求变得非常多,这时该池就会尝试运行更多的线程来清理;这时第二个节流阀—最大线程数就起作用了。

返回类ExecutorService extends Executor

new类ThreadPoolExecutor extends AbstractExecutorService implements ExecutorService extends Executor

本文参与 腾讯云自媒体同步曝光计划,分享自作者个人站点/博客。
原始发表:2021-06-27 ,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 作者个人站点/博客 前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
暂无评论
推荐阅读
编辑精选文章
换一批
java线程池(一):java线程池基本使用及Executors
在前面学习线程组的时候就提到过线程池。实际上线程组在我们的日常工作中已经不太会用到,但是线程池恰恰相反,是我们日常工作中必不可少的工具之一。现在开始对线程池的使用,以及底层ThreadPoolExecutor的源码进行分析。
冬天里的懒猫
2020/09/17
1.4K0
java线程池(一):java线程池基本使用及Executors
【Java 线程池】Java 创建线程池的正确姿势: Executors 和 ThreadPoolExecutor 详解
Executors的创建线程池的方法,创建出来的线程池都实现了ExecutorService接口。常用方法有以下几个:
一个会写诗的程序员
2020/06/03
37.5K0
线程池运行任务后阻塞问题分析
java.util.concurrent.Executors#newFixedThreadPool(int)
明明如月学长
2021/08/31
2K0
线程池运行任务后阻塞问题分析
多线程(三) | 彻底搞懂线程池-ThreadPoolExecutor
上篇文章我们讲解了Executors这个线程池工厂类的用法,介绍了它所提供的一些常用线程池工具的用法,今天我们来继续深入学习。
一缕82年的清风
2022/05/10
9320
多线程(三) | 彻底搞懂线程池-ThreadPoolExecutor
线程池的创建和使用
这里需要注意,线程的数量是固定的,但是队列大小是无界的(Integer.MAX_VALUE足够大,大到可以任务无界。)
IT云清
2019/01/22
1.1K0
关于禁止使用Executors创建线程池的分析
线程池不允许使用Executors去创建,而是通过ThreadPoolExecutor的方式,这样的处理方式让写的同学更加明确线程池的运行规则,规避资源耗尽的风险。 注意,这里的重点是 不允许。而不是不建议。可见该规范 背后都是血淋淋的生产事故。
冬天里的懒猫
2020/08/04
1.6K0
关于禁止使用Executors创建线程池的分析
使用 Executors,ThreadPoolExecutor,创建线程池,源码分析理解
之前创建线程的时候都是用的 newCachedThreadPoo,newFixedThreadPool,newScheduledThreadPool,newSingleThreadExecutor 这四个方法。 当然 Executors 也是用不同的参数去 new ThreadPoolExecutor 实现的,本文先分析前四种线程创建方式,后在分析 new ThreadPoolExecutor 创建方式 使用 Executors 创建线程池 1.newFixedThreadPool() 由于使用了Link
程序员鹏磊
2018/02/09
6860
JAVA线程池的几种使用方式以及线程同步详解
volatile还有一个重要的作用是禁止指令重排序优化,举个例子 下面代码中如果initFlag不用volatile修饰的话,就不能保证程序在运行的时候numOne,numTwo,numThree,str的赋值一定在initFlag之前,多线程并发访问的时候就会有可能出现initFlag为true,但是前面的赋值并没有完成
yingzi_code
2019/08/31
1.8K0
关于Java多线程的一些常考知识点
前言 Java多线程也是面试中经常会提起到的一个点。面试官会问:实现多线程的两种方式以及区别,死锁发生的4个条件以及如何避免发生死锁,死锁和活锁的区别,常见的线程池以及区别,怎么理解有界队列与无界队
用户2032165
2018/06/05
9920
Java线程池参数分析「建议收藏」
实际上Executors创建的线程池实例最终都是通过实例化配置不同参数的ThreadPoolExecutor对象。
全栈程序员站长
2022/09/06
3610
Java线程池参数分析「建议收藏」
Executors Java编程规范插件提示手动创建线程池的解决办法
https://howtodoinjava.com/core-java/multi-threading/when-to-use-countdownlatch-java-concurrency-example-tutorial/
明明如月学长
2021/08/27
5690
线程池之ThreadPoolExecutor使用
知道了各个参数的作用后,我们开始构造符合我们期待的线程池。首先看JDK给我们预定义的几种线程池:
用户6182664
2020/05/07
4290
线程池之ThreadPoolExecutor使用
如果对这些参数作用有疑惑的请看 ThreadPoolExecutor概述。 知道了各个参数的作用后,我们开始构造符合我们期待的线程池。首先看JDK给我们预定义的几种线程池:
Vincent-yuan
2021/07/13
4550
Java--线程池
两者相差的时间为100倍以上,这边使用的线程池回头再来看,由这个例子可以看出,线程的创建到运行过程,会花费大量的cpu时间,争对需要大量并发的操作,使用线程池可以大大提高程序性能
aruba
2021/12/06
3500
Java--线程池
Java 多线程(7)----线程池(下)
在上篇文章:Java 多线程—线程池(上) 中我们看了一下 Java 中的阻塞队列,我们知道阻塞队列是一种可以对线程进行阻塞控制的队列,并且在前面我们也使用了阻塞队列来实现 生产者-消费者模型 。在文章最后,我们还看了一下 Future 接口和其中对应的方法,如果你对这些不熟悉,建议先去看一下上一篇文章。有了前面的知识作为基础之后,我们来正式看一下 Java 中的线程池。
指点
2019/01/18
5570
Java 多线程(7)----线程池(下)
Java并发——线程池(八)
线程池(Thread Pool)是一种基于“池化”思想管理线程的工具,经常出现在多线程服务器中。通过创建一定数量的线程,让这些线程处于就绪状态来提高系统响应速度,在线程使用完成后归还到线程池来达到重复利用的目标,从而降低系统资源的消耗。
翰墨飘香
2024/05/15
1510
一篇搞懂线程池
在上一篇文章《spring boot使用@Async异步任务》中我们了解了使用@Async的异步任务使用,在这篇文章中我们将学习使用线程池来创建异步任务的线程。
小森啦啦啦
2019/07/14
6900
JUC内置线程池
整个线程池表现为线程数会根据任务量不断增长,没有上限,当任务执行完毕,空闲 1分钟后释放线程。 适合任务数比较密集,但每个任务执行时间较短的情况。
兜兜转转
2023/03/08
1800
JUC内置线程池
Java中executors提供的的4种线程池
jdk中关于线程池一个比较核心的类是ThreadPoolExecutor,先来看一下他的实现.
呼延十
2019/06/26
1.2K0
ThreadPoolExecutor 线程池的源码解析
  上一篇从整体上介绍了Executor接口,从上一篇我们知道了Executor框架的最顶层实现是ThreadPoolExecutor类,Executors工厂类中提供的newScheduledThreadPool、newFixedThreadPool、newCachedThreadPool方法
小勇DW3
2018/08/30
4430
相关推荐
java线程池(一):java线程池基本使用及Executors
更多 >
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档
本文部分代码块支持一键运行,欢迎体验
本文部分代码块支持一键运行,欢迎体验