Executors工厂类详解

简介: Java提供了多种线程池创建方法:`newFixedThreadPool` 创建固定数量线程,适用于并发稳定场景;`newCachedThreadPool` 可缓存线程,适合短时高并发任务;`newSingleThreadExecutor` 保证任务顺序执行;`newScheduledThreadPool` 支持定时与周期性任务。不同线程池适应不同业务需求,合理选择可提升系统性能与资源利用率。

public static ExecutorService newFixedThreadPool(int nThreads)


使用这个方法会产生这样一个线程池:线程池最多会保持nThreads个线程处于活动状态,如果当前所有任务都处于活动状态,那么新提交的任务会被添加到任务阻塞队列中去。总结一下就是:使用固定大小的线程池,并发数是固定的。


* 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}.

public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory)


相比于newFixedThreadPool(int nThreads), 你可以使用这个方法来传递你自己的线程工厂,线程工厂是用来干嘛的?就是用来生成线程的,你可以使用线程工厂做一些个性化的线程特性定制。


public static ExecutorService newWorkStealingPool(int parallelism)

在了解或者使用这个方法之前,你你该对java的Fork/Join并行框架有一些了解,如果你想要快速了解一下该部分的内容,可以参考这篇文章:Java Fork/Join并行框架。


从名字上我们就知道这个方法生产出来的线程池具有某种“小偷”的行为,在Fork/Join里面,线程的工作模式为“盗窃算法”,也就是在自己的任务队列消费完了之后不是进入等到状态,而是会主动去偷窃别的线程的任务来做,其实是没有一种奖励机制来鼓励这些线程去帮助别的线程去消费任务的,所以可以认为这些线程都是好人,都为了快速完成任务协调作战。这种工作方式的重点在于,每个线程都将有一个任务队列,线程之间通过“偷窃”的方式互相帮助来完成任务的消费。

可以看下这个方法的实现:


return new ForkJoinPool(parallelism, ForkJoinPool.defaultForkJoinWorkerThreadFactory, null, true);


可以发现,这个方法不是使用我们在第一篇文章中分析了ThreadPoolExecutor来生成线程池。而是使用了ForkJoinPool,也就是Fork/Join里面的线程池,关于ForkJoinPool更为深入的分析不再本文的涉及范围内,你只要知道Fork/Join框架的一般运行原理就可以了,下面的描述可以帮助你决策你是否需要该方法提供的线程池来工作:


* 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.

public static ExecutorService newWorkStealingPool()


参考newWorkStealingPool(int parallelism)。


public static ExecutorService newSingleThreadExecutor()

下面是对该方法的描述:


* 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.


可以从方法的名字上知道,该方法产生的线程池仅仅有一个Worker,任何时刻都将只有一个Worker在工作,添加的任务有很大概率被放在阻塞任务队列中等待执行。这些任务会被顺序执行,这个方法的返回值其实是对ThreadPoolExecutor的一层包装,下面的代码展示了最终执行任务的类:


static class DelegatedExecutorService extends AbstractExecutorService {
    private final ExecutorService e;
    DelegatedExecutorService(ExecutorService executor) { e = executor; }
    public void execute(Runnable command) { e.execute(command); }
    public void shutdown() { e.shutdown(); }
    public List<Runnable> shutdownNow() { return e.shutdownNow(); }
    public boolean isShutdown() { return e.isShutdown(); }
    public boolean isTerminated() { return e.isTerminated(); }
    public boolean awaitTermination(long timeout, TimeUnit unit)
        throws InterruptedException {
        return e.awaitTermination(timeout, unit);
    }
    public Future<?> submit(Runnable task) {
        return e.submit(task);
    }
    public <T> Future<T> submit(Callable<T> task) {
        return e.submit(task);
    }
    public <T> Future<T> submit(Runnable task, T result) {
        return e.submit(task, result);
    }
    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
        throws InterruptedException {
        return e.invokeAll(tasks);
    }
    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                                         long timeout, TimeUnit unit)
        throws InterruptedException {
        return e.invokeAll(tasks, timeout, unit);
    }
    public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
        throws InterruptedException, ExecutionException {
        return e.invokeAny(tasks);
    }
    public <T> T invokeAny(Collection<? extends Callable<T>> tasks,
                           long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException {
        return e.invokeAny(tasks, timeout, unit);
    }
}


从上面的代码可以看出,这个类其实就是使用了构造时传递的参数e来完成,更像是代理。而e是什么?看下面的代码:


ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>())

其实就是一个只有一个线程的ThreadPoolExecutor。

public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory)
参考newSingleThreadExecutor(),多了一个线程工厂参数

public static ExecutorService newCachedThreadPool()

首先看它的方法体内容:


return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());


可以看到,核心线程数量为0,而上限为Integer.MAX_VALUE,而且keepAliveTime为60秒,那么这个线程池的工作模式为:只要有任务呗提交,而且当前没有空闲的线程可用,那么就会创建一个新的Worker来工作,一个线程工作完了之后会缓存(idle)60秒,如果60秒之内有新的任务提交,则会被唤醒进入工作模式,否则60秒后就会被回收。可以参考下面的的描述:


* 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.


从描述上,我们可以想到,其实这种类型的线程池比较适合于短期高流量的场景,也就是我们所说的“秒杀”场景,在那样的场景下,需要的线程数量较多,那么使用该类型的线程池可以满足,而且该线程池还有自动收缩的功能,在不需要那么多线程的时候,会自动回收线程,释放资源。


public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory)

参考newCachedThreadPool()。

public static ScheduledExecutorService newSingleThreadScheduledExecutor()


只有一个线程的调度线程池,类似于newSingleThreadExecutor,但是该方法生产的线程池具备调度功能,下面是对该方法的描述:


* Creates a single-threaded executor that can schedule commands
     * to run after a given delay, or to execute periodically.
     * (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 newScheduledThreadPool(1)} the returned executor is
     * guaranteed not to be reconfigurable to use additional threads.

public static ScheduledExecutorService newSingleThreadScheduledExecutor(ThreadFactory threadFactory)

参考newSingleThreadExecutor和newSingleThreadScheduledExecutor。


public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize)


参考newFixedThreadPool,但是返回类型不一样。


public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize, ThreadFactory threadFactory)


参考newFixedThreadPool。


通过上面的分析,我们应该对java线程池的理解更为深入,再次说明,第五节是对前面四节内容的补充,你应该首先前四节之后再来阅读第五节,那样内容上更完整,但是单独阅读本文一样具备独立性,但是收获肯定没有同时阅读那样多。

相关文章
|
人工智能 vr&ar
AIGC领域又有大动作了!AI视频“黑马”Morph Studio 来袭,1080P高清画质视频免费生成!
AIGC领域又有大动作了!AI视频“黑马”Morph Studio 来袭,1080P高清画质视频免费生成!
838 2
|
SQL 存储 druid
Minerva -- Airbnb 的大规模数据指标系统 Part 3
Minerva -- Airbnb 的大规模数据指标系统 Part 3
1148 0
Minerva -- Airbnb 的大规模数据指标系统 Part 3
|
10月前
|
监控 NoSQL 中间件
阿里云服务器通用算力型u2a实例性能、适用场景,最新收费标准与活动价格参考
本文聚焦阿里云2026年通用算力型u2a实例,该实例基于AMD CPU平台打造,支持跨平台热迁移,适配多代际CPU处理器。其搭载AMD EPYC处理器,在MySQL、Redis等数据库应用中性能提升超50%,适配中小型网站、应用服务器及中间件服务等多场景。文章详细解析了u2a实例的规格配置、按量付费与包年包月价格体系,以及2026年新用户2.5折优惠活动,并提供业务场景匹配建议与成本优化策略,助力中小企业及个人用户低成本高效上云。
630 1
|
2天前
|
人工智能 API Apache
Qwen3.8抢先体验!大家要心理准备了,最新版发布并开源了?
Qwen3.8系列已于2026年8月起全线开源,涵盖Max(2.4T)、27B(Apache 2.0)、Flash及Flash-Next四版本,支持本地部署与商用(部分需授权)。Qwen4仍在训练,尚未发布。
285 0
|
6月前
|
供应链 安全 网络协议
从Silver Fox新变种看2026年网络钓鱼的攻防进化
2026年初,德企Hexastrike披露“银狐”团伙利用AtlasCross RAT木马,通过伪造Zoom、Signal等知名软件域名实施精准钓鱼。攻击融合拼错域名、盗用代码签名证书、无文件内存注入及针对性反杀软技术,严重威胁亚洲商务与财务人员安全。
543 3
|
3月前
|
缓存 开发工具 git
DeepSeek Harness 更新失败怎么办?DSH plugin 升级报错排查与版本回滚指南
DeepSeek Harness 更新失败先分安装方式排查:npx 报错多为网络与 npm 缓存问题,源码构建遇 git pull 冲突要先解决本地改动;失败后可回滚到旧版本。DSH plugin 升级报错可在 DSH Plugin Hub 内置插件中心重新安装。
828 0
|
8月前
|
安全 机器人 Linux
零基础零技术|OpenClaw阿里云+本地多系统部署+千问/Coding Plan API配置+钉钉接入全指南
OpenClaw(原Clawdbot,昵称“龙虾”)是一款开源AI自动化代理框架,核心功能是通过自然语言指令拆解,调用各类技能完成文件管理、办公协同、数据抓取、代码生成等重复性任务,无需手动编写脚本,大幅提升个人与轻量团队的工作效率。其核心依赖大模型提供推理能力,自身不具备独立的自然语言理解与任务规划能力,因此部署流程、API配置以及办公工具接入,成为零基础用户落地OpenClaw的核心痛点。
1170 0
|
6月前
|
人工智能 监控 测试技术
AI 测试用例审核 Skill:把用例评审从“凭经验”变成“可评分”
本文介绍一种AI驱动的测试用例审核Skill,将资深测试负责人的评审经验封装为可复用、可量化、可批量执行的标准能力。它能自动检查逻辑完整性、预期明确性、前置条件、PRD覆盖度及边界异常,逐条评分、定位问题、给出修改建议,助力团队提升用例质量、统一评审标准、加速新人成长。
|
7月前
|
存储 人工智能 自然语言处理
Claude Opus 4.7 系统 Prompt 泄露:其中的10 个核心设计决策解读
刚发布的Claude 4.7 Prompt遭泄露,揭示其核心设计哲学:不只追求“更聪明”,更强调“自我约束”。它将心理重构、过度礼貌、工具犹豫等常见AI倾向视为风险信号,通过情感化规则、动态安全升级、隐式上下文记忆等10项机制,系统性抑制自身失败模式——真正目标是让模型清醒认知并管控自身的不可靠性。
404 2
Claude Opus 4.7 系统 Prompt 泄露:其中的10 个核心设计决策解读
|
机器学习/深度学习 数据采集 人工智能
【机器学习】怎样检测到线性回归模型中的过拟合?
【5月更文挑战第17天】【机器学习】怎样检测到线性回归模型中的过拟合?