目录
先理解线程池到底有什么作用
先看ThreadPoolExecutor类的一系列代表状态的
该类一共给出了五种状态
接下来是线程池的几个有关工作线程的变量
下面是线程池默认的线程工厂的生产线程方法
我们可以先看我们最常调用的execute()方法
下面是addWorker()的具体方法
Worker类作为线程池的内部类
run()方法实现的很简单
下面是getTask()的实现
在runWorker()方法的最后
首页 Java java教程 Java线程池execute()方法怎么用

Java线程池execute()方法怎么用

May 10, 2023 pm 11:46 PM
java execute

先理解线程池到底有什么作用

* Thread pools address two different problems: they usually
* provide improved performance when executing large numbers of
* asynchronous tasks, due to reduced per-task invocation overhead,
* and they provide a means of bounding and managing the resources,
* including threads, consumed when executing a collection of tasks.
* Each {@code ThreadPoolExecutor} also maintains some basic
* statistics, such as the number of completed tasks.
登录后复制

线程池处理了两个不同的问题,线程池通过减少线程正式调用之前的开销来给大量异步任务更优秀的表现,与此同时给出了一系列绑定管理任务线程的一种手段。每个线程池都包含了一些基本信息,比如内部完成的任务数量。

先看ThreadPoolExecutor类的一系列代表状态的

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;
 
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;
 
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作为AtomicInteger类存放了类中的两种信息,在其中由高3位来保存线程池的状态,后29位来保存此时线程池中的Woker类线程数量(由此可知,线程池中的线程数量最高可以接受大约在五亿左右)。由此可见给出的runStateOf()和workerCountOf()方法分别给出了查看线程状态和线程数量的方法。

该类一共给出了五种状态

让我们看作者给出的注释

*   RUNNING:  Accept new tasks and process queued tasks
*   SHUTDOWN: Don&#39;t accept new tasks, but process queued tasks
*   STOP:     Don&#39;t accept new tasks, don&#39;t process queued tasks,
*             and interrupt in-progress tasks
*   TIDYING:  All tasks have terminated, workerCount is zero,
*             the thread transitioning to state TIDYING
*             will run the terminated() hook method
*   TERMINATED: terminated() has completed
登录后复制
  • RUNNING状态可以接受新进来的任务,同时也会执行队列里的任务。

  • SHUTDOWN 状态已经不会再接受新任务,但仍旧会处理队列中的任务。

  • STOP状态在之前的基础上,不会处理队列中的人物,在执行的任务也会直接被打断。

  • TIDYING状态在之前的基础上,所有任务都已经终止,池中的Worker线程都已经为0,也就是stop状态在清理完所有工作线程之后就会进入该状态,同时在shutdown状态在队列空以及工作线程清理完毕之后也会直接进入这个阶段,这一阶段会循环执行terminated()方法。

  • TERMINATED 状态作为最后的状态,在之前的基础上terminated()方法也业已执行完毕,才会从上个状态进入这个状态,代表线程池已经完全停止。

由于线程池的状态都是通过AtomicInteger来保存的,可以通过比较的方式简单的得到当前线程状态。

private final BlockingQueue<Runnable> workQueue; 
private final ReentrantLock mainLock = new ReentrantLock(); 
private final HashSet<Worker> workers = new HashSet<Worker>(); 
private final Condition termination = mainLock.newCondition(); 
private int largestPoolSize; 
private long completedTaskCount; 
private volatile ThreadFactory threadFactory; 
private volatile RejectedExecutionHandler handler; 
private volatile long keepAliveTime; 
private volatile boolean allowCoreThreadTimeOut; 
private volatile int corePoolSize; 
private volatile int maximumPoolSize;
登录后复制

接下来是线程池的几个有关工作线程的变量

  • corePoolSize表示线程池中允许存活最少的工作线程数量,但值得注意的是如果allowCoreThreadTimeOut一旦设置true(默认false),每个线程的存活时间只有keepAliveTime也就是说在allowCoreThreadTimeOut为true的时候,该线程池最小的工作线程数量为0;maximumPoolSize代表线程池中最大的工作线程数量。

  • keepAliveTime为线程池中工作线程数量大于corePoolSize时,每个工作线程的在等待工作时最长的等待时间。

  • workQueue作为线程池的任务等待队列,这个将在接下来的execute()里详细解释。

  • Workers作为存放线程池中存放工作线程的容器。

  • largestPoolSize用来记录线程池中存在过的最大的工作线程数量。

  • completedTaskCount用来记录线程池完成的任务的总数。

  • Handler作为线程池中在不能接受任务的时候的拒绝策略,我们可以实现自己的拒绝策略,在实现了RejectedExecutionHandler接口的前提下。下面是线程池的默认拒绝策略,

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    throw new RejectedExecutionException("Task " + r.toString() +
                                         " rejected from " +
                                         e.toString());
}
登录后复制

threadFactory作为线程池生产线程的工厂类

下面是线程池默认的线程工厂的生产线程方法

public Thread newThread(Runnable r) {
    Thread t = new Thread(group, r,
                          namePrefix + threadNumber.getAndIncrement(),
                          0);
    if (t.isDaemon())
        t.setDaemon(false);
    if (t.getPriority() != Thread.NORM_PRIORITY)
        t.setPriority(Thread.NORM_PRIORITY);
    return t;
}
登录后复制

我们可以先看我们最常调用的execute()方法

public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
 
    int c = ctl.get();
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        if (! isRunning(recheck) && remove(command))
            reject(command);
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    else if (!addWorker(command, false))
        reject(command);
}
登录后复制

execute()内部的调用逻辑非常清晰。

如果当前线程池的工作线程数量小于corePoolSize,那么直接调用addWoker(),来添加工作线程。

下面是addWorker()的具体方法

private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);
        if (rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN &&
               firstTask == null &&
               ! workQueue.isEmpty()))
            return false;
 
        for (;;) {
            int wc = workerCountOf(c);
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            if (compareAndIncrementWorkerCount(c))
                break retry;
            c = ctl.get();  // Re-read ctl
            if (runStateOf(c) != rs)
                continue retry;
        }
    }
 
    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        final ReentrantLock mainLock = this.mainLock;
        w = new Worker(firstTask);
        final Thread t = w.thread;
        if (t != null) {
            mainLock.lock();
            try {
                int c = ctl.get();
                int rs = runStateOf(c);
 
                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) {
                    if (t.isAlive())                         
           throw new IllegalThreadStateException();
                    workers.add(w);
                    int s = workers.size();
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
            if (workerAdded) {
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        if (! workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}
登录后复制

这段方法比较长,但整体的逻辑还是清晰的。

首先判断当前线程池的状态,如果已经状态不是shutdown或者running,或者已经为shutdown但是工作队列已经为空,那么这个时候直接返回添加工作失败。接下来是对线程池线程数量的判断,根据调用时的core的值来判断是跟corePoolSize还是 maximumPoolSize判断。

在确认了线程池状态以及线程池中工作线程数量之后,才真正开始添加工作线程。

新建立一个worker类(线程池的内部类,具体的工作线程),将要执行的具体线程做为构造方法中的参数传递进去,接下来将其加入线程池的工作线程容器workers,并且更新工作线程最大量,最后调用worker工作线程的start()方法,就完成了工作线程的建立与启动。

让我们回到execute()方法,如果我们在一开始的线程数量就大于corePoolSize,或者我们在调用addworker()方法的过程中出现了问题导致添加工作线程数量失败,那么我们会继续执行接下来的逻辑。

在判断完毕线程池的状态后,则会将任务通过workQueue.offer())方法试图加进任务队列。Offer()方法的具体实现会根据在线程池构造方法中选取的任务队列种类而产生变化。

但是如果成功加入了任务队列,仍旧需要注意判断如果线程池的状态如果已经不是running那么会拒绝执行这一任务并执行相应的拒绝策略。在最后需要记得成功加入队列成功后如果线程池中如果已经没有了工作线程,需要重新建立一个工作线程去执行仍旧在任务队列中等待执行的任务。

如果在之前的前提下加入任务队列也失败了(比如任务队列已满),则会在不超过线程池最大线程数量的前提下建立一个工作线程来处理。

如果在最后的建立工作线程也失败了,那么我们只有很遗憾的执行任务的拒绝策略了。

在之前的过程中我们建立了工作线程Worker()类,那么我们现在看看worker类的内部实现,也可以说是线程池的核心部分。

Worker类作为线程池的内部类

接下来是Worker()类的成员

final Thread thread;
 
Runnable firstTask;
 
volatile long completedTasks;
登录后复制
  • thread作为worker的工作线程空间,由线程池中所设置的线程工厂生成。

  • firstTask则是worker在构造方法中所接受到的所要执行的任务。

  • completedTasks作为该worker类所执行完毕的任务总数。

接下来我们可以看最重要的,也就是我们之前建立完Worker类之后立马调用的run()方法了

public void run() {
    runWorker(this);
}
登录后复制

run()方法实现的很简单

我们可以继续追踪下去

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); 
    boolean completedAbruptly = true;
    try {
        while (task != null || (task = getTask()) != null) {
            w.lock();
            if ((runStateAtLeast(ctl.get(), STOP) ||
                 (Thread.interrupted() &&
                  runStateAtLeast(ctl.get(), STOP))) &&
                !wt.isInterrupted())
                wt.interrupt();
            try {
                beforeExecute(wt, task);
                Throwable thrown = null;
                try {
                    task.run();
                } catch (RuntimeException x) {
                    thrown = x; throw x;
                } catch (Error x) {
                    thrown = x; throw x;
                } catch (Throwable x) {
                    thrown = x; throw new Error(x);
                } finally {
                    afterExecute(task, thrown);
                }
            } finally {
                task = null;
                w.completedTasks++;
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        processWorkerExit(w, completedAbruptly);
    }
}
登录后复制

如果这个worker还没有执行过在构造方法就传入的任务,那么在这个方法中,会直接执行这一任务,如果没有,则会尝试去从任务队列当中去取的新的任务。

但是在真正调用任务之前,仍旧会判断线程池的状态,如果已经不是running亦或是shutdwon,则会直接确保线程被中断。如果没有,将会继续执行并确保不被中断。

接下来可见,我们所需要的任务,直接在工作线程中直接以run()方式以非线程的方式所调用,这里也就是我们所需要的任务真正执行的地方。

在执行完毕后,工作线程的使命并没有真正宣告段落。在while部分worker仍旧会通过getTask()方法试图取得新的任务。

下面是getTask()的实现

private Runnable getTask() {
    boolean timedOut = false; 
    retry:
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);
 
               if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            decrementWorkerCount();
            return null;
        }
 
        boolean timed;            
        for (;;) {
            int wc = workerCountOf(c);
            timed = allowCoreThreadTimeOut || wc > corePoolSize;
 
            if (wc <= maximumPoolSize && ! (timedOut && timed))
                break;
            if (compareAndDecrementWorkerCount(c))
                return null;
            c = ctl.get();  
            if (runStateOf(c) != rs)
                continue retry;
        }
 
        try {
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
            if (r != null)
                return r;
            timedOut = true;
        } catch (InterruptedException retry) {
            timedOut = false;
        }
    }
}
登录后复制

首先仍旧会判断线程池的状态是否是running还是shutdown以及stop状态下队列是否仍旧有需要等待执行的任务。如果状态没有问题,则会跟据allowCoreThreadTimeOut和corePoolSize的值通过对前面这两个属性解释的方式来选择从任务队列中获得任务的方式(是否设置timeout)。其中的timedOut保证了确认前一次试图取任务时超时发生的记录,以确保工作线程的回收。

在runWorker()方法的最后

调用了processWorkerExist()方法来执行工作线程的回收。

private void processWorkerExit(Worker w, boolean completedAbruptly) {
    if (completedAbruptly) 
        decrementWorkerCount();
 
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        completedTaskCount += w.completedTasks;
        workers.remove(w);
    } finally {
        mainLock.unlock();
    }
 
    tryTerminate(); 
    int c = ctl.get();
    if (runStateLessThan(c, STOP)) {
        if (!completedAbruptly) {
            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
            if (min == 0 && ! workQueue.isEmpty())
                min = 1;
            if (workerCountOf(c) >= min)
                return; 
        }
        addWorker(null, false);
    }
}
登录后复制

在这一方法中,首先确保已经重新更新了线程池中工作线程的数量,之后从线程池中的工作线程容器移去当前工作线程,并且将完成的任务总数加到线程池的任务总数当中。

在最后仍旧要确保线程池中依旧存在大于等于最小线程数量的工作线程数量存在,如果没有,则重新建立工作线程去等待处理任务队列中任务。

以上是Java线程池execute()方法怎么用的详细内容。更多信息请关注PHP中文网其他相关文章!

本站声明
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系admin@php.cn

热AI工具

Undresser.AI Undress

Undresser.AI Undress

人工智能驱动的应用程序,用于创建逼真的裸体照片

AI Clothes Remover

AI Clothes Remover

用于从照片中去除衣服的在线人工智能工具。

Undress AI Tool

Undress AI Tool

免费脱衣服图片

Clothoff.io

Clothoff.io

AI脱衣机

Video Face Swap

Video Face Swap

使用我们完全免费的人工智能换脸工具轻松在任何视频中换脸!

热工具

记事本++7.3.1

记事本++7.3.1

好用且免费的代码编辑器

SublimeText3汉化版

SublimeText3汉化版

中文版,非常好用

禅工作室 13.0.1

禅工作室 13.0.1

功能强大的PHP集成开发环境

Dreamweaver CS6

Dreamweaver CS6

视觉化网页开发工具

SublimeText3 Mac版

SublimeText3 Mac版

神级代码编辑软件(SublimeText3)

Java 中的史密斯数 Java 中的史密斯数 Aug 30, 2024 pm 04:28 PM

Java 史密斯数指南。这里我们讨论定义,如何在Java中检查史密斯号?带有代码实现的示例。

Java Spring 面试题 Java Spring 面试题 Aug 30, 2024 pm 04:29 PM

在本文中,我们保留了最常被问到的 Java Spring 面试问题及其详细答案。这样你就可以顺利通过面试。

突破或从Java 8流返回? 突破或从Java 8流返回? Feb 07, 2025 pm 12:09 PM

Java 8引入了Stream API,提供了一种强大且表达力丰富的处理数据集合的方式。然而,使用Stream时,一个常见问题是:如何从forEach操作中中断或返回? 传统循环允许提前中断或返回,但Stream的forEach方法并不直接支持这种方式。本文将解释原因,并探讨在Stream处理系统中实现提前终止的替代方法。 延伸阅读: Java Stream API改进 理解Stream forEach forEach方法是一个终端操作,它对Stream中的每个元素执行一个操作。它的设计意图是处

Java 中的时间戳至今 Java 中的时间戳至今 Aug 30, 2024 pm 04:28 PM

Java 中的时间戳到日期指南。这里我们还结合示例讨论了介绍以及如何在java中将时间戳转换为日期。

Java程序查找胶囊的体积 Java程序查找胶囊的体积 Feb 07, 2025 am 11:37 AM

胶囊是一种三维几何图形,由一个圆柱体和两端各一个半球体组成。胶囊的体积可以通过将圆柱体的体积和两端半球体的体积相加来计算。本教程将讨论如何使用不同的方法在Java中计算给定胶囊的体积。 胶囊体积公式 胶囊体积的公式如下: 胶囊体积 = 圆柱体体积 两个半球体体积 其中, r: 半球体的半径。 h: 圆柱体的高度(不包括半球体)。 例子 1 输入 半径 = 5 单位 高度 = 10 单位 输出 体积 = 1570.8 立方单位 解释 使用公式计算体积: 体积 = π × r2 × h (4

PHP与Python:了解差异 PHP与Python:了解差异 Apr 11, 2025 am 12:15 AM

PHP和Python各有优势,选择应基于项目需求。1.PHP适合web开发,语法简单,执行效率高。2.Python适用于数据科学和机器学习,语法简洁,库丰富。

PHP:网络开发的关键语言 PHP:网络开发的关键语言 Apr 13, 2025 am 12:08 AM

PHP是一种广泛应用于服务器端的脚本语言,特别适合web开发。1.PHP可以嵌入HTML,处理HTTP请求和响应,支持多种数据库。2.PHP用于生成动态网页内容,处理表单数据,访问数据库等,具有强大的社区支持和开源资源。3.PHP是解释型语言,执行过程包括词法分析、语法分析、编译和执行。4.PHP可以与MySQL结合用于用户注册系统等高级应用。5.调试PHP时,可使用error_reporting()和var_dump()等函数。6.优化PHP代码可通过缓存机制、优化数据库查询和使用内置函数。7

创造未来:面向零基础的 Java 编程 创造未来:面向零基础的 Java 编程 Oct 13, 2024 pm 01:32 PM

Java是热门编程语言,适合初学者和经验丰富的开发者学习。本教程从基础概念出发,逐步深入讲解高级主题。安装Java开发工具包后,可通过创建简单的“Hello,World!”程序实践编程。理解代码后,使用命令提示符编译并运行程序,控制台上将输出“Hello,World!”。学习Java开启了编程之旅,随着掌握程度加深,可创建更复杂的应用程序。

See all articles