全部标签
写点什么
OpenHarmony算法元宇宙MySQL移动开发学习方法学习方法Web3.0高效工作数据库Python音视频前端AI大数据团队管理程序员运维深度思考低代码redisgolang微服务架构flutter 查看更多

Java 线程池 ThreadPoolExecutor 源码解析

作者：快乐非自愿限量之名
  • 2023-12-29
    福建

  • 本文字数：10494 字

    阅读完需：约 34 分钟

Java 线程池 ThreadPoolExecutor 源码解析


1.ThreadPoolExecutor 的构造实现


以 jdk8 为准，常说线程池有七大参数，通常而言，有四个参数是比较重要的


  public ThreadPoolExecutor(int corePoolSize,                              int maximumPoolSize,                              long keepAliveTime,                              TimeUnit unit,                              BlockingQueue<Runnable> workQueue) {        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,             Executors.defaultThreadFactory(), defaultHandler);    }
复制代码


  • corePoolSize:核心线程数,具体含义理解代码

  • maximumPoolSize:最大线程数

  • keepAliveTime:线程空闲的存活时间

  • unit:时间单位

  • BlockingQueue:阻塞队列,用来保存等待执行的任务


接下来去看完整参数的构造实现:


public ThreadPoolExecutor(int corePoolSize,                              int maximumPoolSize,                              long keepAliveTime,                              TimeUnit unit,                              BlockingQueue<Runnable> workQueue,                              ThreadFactory threadFactory,                              RejectedExecutionHandler handler) {        if (corePoolSize < 0 ||            maximumPoolSize <= 0 ||            maximumPoolSize < corePoolSize ||            keepAliveTime < 0)            throw new IllegalArgumentException();        if (workQueue == null || threadFactory == null || handler == null)            throw new NullPointerException();        this.acc = System.getSecurityManager() == null ?                null :                AccessController.getContext();        this.corePoolSize = corePoolSize;        this.maximumPoolSize = maximumPoolSize;        this.workQueue = workQueue;        this.keepAliveTime = unit.toNanos(keepAliveTime);        this.threadFactory = threadFactory;        this.handler = handler;    }
复制代码


  • ThreadFactory:线程工厂,用来创造线程


  • RejectedExecutionHandler:拒绝策略


  1. 如果核心线程数等其他参数非法化就会抛出相应的异常


 if (corePoolSize < 0 ||            maximumPoolSize <= 0 ||            maximumPoolSize < corePoolSize ||            keepAliveTime < 0)            throw new IllegalArgumentException();        if (workQueue == null || threadFactory == null || handler == null)            throw new NullPointerException();
复制代码


  1. 之后进行初始化赋值


 this.acc = System.getSecurityManager() == null ?                null :                AccessController.getContext();        this.corePoolSize = corePoolSize;        this.maximumPoolSize = maximumPoolSize;        this.workQueue = workQueue;        this.keepAliveTime = unit.toNanos(keepAliveTime);        this.threadFactory = threadFactory;        this.handler = handler;
复制代码


  • 注:acc 是一个成员变量,用来管理线程池中线程的访问控制上下文,其实现类是AccessControlContext

2.线程池的执行 execute


public void execute(Runnable command) {        if (command == null)            throw new NullPointerException();                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);    }
复制代码


总共大致分为三步:要想理解线程池的执行,要先去理解控制字段其具体含义


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;
// runState is stored in the high-order bitsprivate 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;
// Packing and unpacking ctlprivate 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; }
复制代码


先声明线程池的五种状态,再看其他字段方法具体执行了其他什么操作


  • RUNNING: -1 << COUNT_BITS，即高 3 位为 111


  • SHUTDOWN: 0 << COUNT_BITS，即高 3 位为 000


  • STOP : 1 << COUNT_BITS，即高 3 位为 001


  • TIDYING : 2 << COUNT_BITS,即高 3 位为 010


  • TERMINATED: 3 << COUNT_BITS,即高 3 位为 011


至于其每种空置状态的具体意义,根据英文释义结合代码具体理解,而非直接理解,通过位移位的操作将高 3 位与低 29 位分离开来,高三位表示此时整个线程池的运行状态,低 29 位表示线程池中线程的数量,再去看 execute 执行过程即可.


int c = ctl.get(); if (workerCountOf(c) < corePoolSize) {            if (addWorker(command, true))                return;            c = ctl.get();        }
复制代码


  • 用于获取此时线程中的线程数,如果小于核心线程数,就添加任务,添加任务成功则返回,失败则重新获取控制字段,addworker后续了解,复杂的东西简单化,理解大致操作思想最为核心.


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);        }
复制代码


  • 根据控制字段 c 去判断线程池的运行状态是否正在运行,如果添加任务成功则不会执行失败,或者说此时线程数有可能已经大于了核心线程数也有可能走到这,所以会将任务添加到阻塞队列中去,然后重新去获得控制字段,再去做校验,如果此时线程池不是正在运行的状态并且删除任务成功,这一步主要是为了防止阻塞队列添加任务成功这个过程,可能线程池不运行了,那么这时候就需要将添加的那个任务删除,并对他执行拒绝策略,又或者是此时线程池中的线程数已经为 0,说明没有线程在工作了,因此添加一个空任务,至于第二个参数在addWorker中在做说明


else if (!addWorker(command, false))            reject(command);
复制代码


  • 字面意思就是添加任务失败,执行拒绝策略,则是为了应对线程池已经到了满负载的状态

3.线程池的任务添加 addworker


private boolean addWorker(Runnable firstTask, boolean core) {        retry:        for (;;) {            int c = ctl.get();            int rs = runStateOf(c);
            // Check if queue empty only if necessary.            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;                // else CAS failed due to workerCount change; retry inner loop            }        }
        boolean workerStarted = false;        boolean workerAdded = false;        Worker w = null;        try {            w = new Worker(firstTask);            final Thread t = w.thread;            if (t != null) {                final ReentrantLock mainLock = this.mainLock;                mainLock.lock();                try {                    // Recheck while holding lock.                    // Back out on ThreadFactory failure or if                    // shut down before lock acquired.                    int rs = runStateOf(ctl.get());
                    if (rs < SHUTDOWN ||                        (rs == SHUTDOWN && firstTask == null)) {                        if (t.isAlive()) // precheck that t is startable                            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;    }
复制代码


代码量很长,但是大致可分为两部分,且逻辑很清晰


  1. 这里使用了标签语法,前半段大致是是否需要添加线程做一系列准备


 retry:        for (;;) {            int c = ctl.get();            int rs = runStateOf(c);
            // Check if queue empty only if necessary.            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;                // else CAS failed due to workerCount change; retry inner loop            }        }
复制代码


  1. 获取控制字段c:这个字段包含了运行状态信息和线程池数量信息,是一个复合字段,而rs则是获取高三位的线程池状态信息


  1. 根据前面线程池状态信息,运行RUNNING值最小,因此判断线程池如果处于非运行的状态,则去判断是否处于关闭的状态,判断第一个任务是否为空,队列不为空,但是由于前面取反操作,其真正含义是:如果线程池的状态不是 SHUTDOWN，或者任务队列为空，或者有待执行的任务，那么就不会拒绝新任务的提交,否则就返回 false,表示添加任务失败


  1. 接下来死循环表示需要去添加执行的任务,首先获取线程池中的线程数,关键的地方在这,如果此时的线程数大于等于容量或者(这里根据传递的参数core来选则比较的目标是核心线程数还是最大线程数),比较失败,则说明超过了接受的范畴,添加任务失败,如果没有失败,则通过底层 CAS 操作使得线程数加 1,然后直接结束调用,跳出循环,,如果 CAS 失败,则说明 ctl 字段受到了变化,此期间有其他任务参与,重新获取此字段,去判断一下重新获取的 ctl 字段和之前的 rs 字段是否相等,这是为了保障多线程情况下出现的一种并发竞争问题导致的线程数发生错乱.


  • 至此,上半部分的核心已经解决,下半部分此时真正去实现任务的添加,通过线程池中的内部类 Worker 去实现


  boolean workerStarted = false;        boolean workerAdded = false;        Worker w = null;        try {            w = new Worker(firstTask);            final Thread t = w.thread;            if (t != null) {                final ReentrantLock mainLock = this.mainLock;                mainLock.lock();                try {                    // Recheck while holding lock.                    // Back out on ThreadFactory failure or if                    // shut down before lock acquired.                    int rs = runStateOf(ctl.get());
                    if (rs < SHUTDOWN ||                        (rs == SHUTDOWN && firstTask == null)) {                        if (t.isAlive()) // precheck that t is startable                            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;    }
复制代码


两个布尔类型暂时不用管,大致猜测意思即可,将 firstask 任务交付给 worker,由 worker 内部的 thread 线程去执行,因此需要去理解 worker 的实现


3.1Worker 内部的工作者


3.11 构造方法实现


Worker(Runnable firstTask) {    setState(-1); // inhibit interrupts until runWorker    this.firstTask = firstTask;    this.thread = getThreadFactory().newThread(this);}
复制代码


接收一个Runnable参数做为任务进行初始化,这里用到了 AQS 的一些实现,然后通过线程工厂创造一个新的线程,赋予给内部的成员变量引用


  • 还有一些锁的一些操作,后续再看


  1. 如果工作者的内部线程已经被创造好,实现就绪,要先获得线程池的互斥锁,然后对接下来的操作进行互斥访问


  1. 重新获取最新的线程池的运行状态,只有当线程池处于运行状态或者处于关闭状态但没有待执行的任务时，才能将新工作线程添加到线程池中,也就是 worker 中去,因此一个 worker 内部具备一个 thread,如果想要实现许多线程去完成线程池的相应操作,需要将 worker 封装成集合,因此线程池内部还有一成员变量:


private final HashSet<Worker> workers = new HashSet<Worker>();
复制代码


这样就确保了每一个worker都是独一无二的,不会重复的,也就意味着每一个线程都不一样.


  1. 而最后一个largestPoolSize则是保留历史的最大线程数的,用来记录,至此就已经添加成功了,只不过此时还没有执行


  1. 之后解锁,用之前标志位workerAdded表示添加成功,然后启动线程,也就是去执行这个任务,再用另一个标志workerStarted表示启动成功.


  1. 最后则是检查是否有什么异常在启动期间,如果没启动成功,则调用addWorkerFiled方法去处理


3.12 addWorkerFailed 方法


private void addWorkerFailed(Worker w) {    final ReentrantLock mainLock = this.mainLock;    mainLock.lock();    try {        if (w != null)            workers.remove(w);        decrementWorkerCount();        tryTerminate();    } finally {        mainLock.unlock();    }}
复制代码


  • 紧接上文,也就是启动失败的话,会将执行任务的 workerremove(底层通过 HashMap 实现键的删除),然后减少线程数,等待一会,这个过程是互斥的,因为牵扯到控制字段


  1. 至此,添加任务如果成功,则进行执行,如果成功开启执行,则成功返回


因此,根据线程池的执行添加流程,大致可以将此过程通过绘图的方式表现出来:



4.工作者的 run 方法是如何运行的


在worker中还有一个方法


  public void run() {            runWorker(this);        }
复制代码


是其执行的具体操作


final void runWorker(Worker w) {        Thread wt = Thread.currentThread();        Runnable task = w.firstTask;        w.firstTask = null;        w.unlock(); // allow interrupts        boolean completedAbruptly = true;        try {            while (task != null || (task = getTask()) != null) {                w.lock();                // If pool is stopping, ensure thread is interrupted;                // if not, ensure thread is not interrupted.  This                // requires a recheck in second case to deal with                // shutdownNow race while clearing interrupt                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);        }    }
复制代码


  • 上锁之前的操作都很容易看懂,处了getTask,这个方法用来获取阻塞队列中的任务,后续再理解


  1. 首先看第一段 if,就是用来查看此时线程池的状态,如果不处于关闭或者运行的状态,或者线程处于中断的状态,则确保线程中断


  1. 接下来是一部分异常和错误的处理以及执行一些前置任务和一些后置任务


  1. 最后完成的任务数加一,解锁,将标志位是否中断,改为 false,表示执行成功.

5.获取任务的 getTask 方法


private Runnable getTask() {    boolean timedOut = false; // Did the last poll() time out?
    for (;;) {        int c = ctl.get();        int rs = runStateOf(c);
        // Check if queue empty only if necessary.        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {            decrementWorkerCount();            return null;        }
        int wc = workerCountOf(c);
        // Are workers subject to culling?        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
        if ((wc > maximumPoolSize || (timed && timedOut))            && (wc > 1 || workQueue.isEmpty())) {            if (compareAndDecrementWorkerCount(c))                return null;            continue;        }
        try {            Runnable r = timed ?                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :                workQueue.take();            if (r != null)                return r;            timedOut = true;        } catch (InterruptedException retry) {            timedOut = false;        }    }}
复制代码


  1. 首先标志位用来判断是否超时，默认情况下不超时，跟之前的参数挂钩，后续再看，然后进入死循环，不断循环去执行后续操作


  1. 获取控制位 c 和 rs 运行状态，之后的 if 操作含义是如果线程池处于关闭的状态或者此时队列为空，就说明没有任务需要处理了，此时让线程池中的线程数减一，返回，另一种情况则是线程池的状态处于关闭状态之上，则说明线程池现在不执行任务了，不需要管队列中是否还有任务存在，则同上减一，返回。


  1. 然后重新获取线程池的线程数，接下来的 time 布尔这个字段有些作用，后面的allowCoreThreadTimeOut是一个控制字段,用来表示线程是否允许超时而返回的一个字段,如:线程池中的核心线程如果因为长时间没有得到任务的滋养,就如同线程之间会发生饥饿一样,因此存在一个字段用来控制超时是否生效.因此如果线程数大于核心线程数或者开启超时控制字段,就说明会执行超时退出.


  1. 接下来的 if 判断是用来执行递减线程数的一个操作,底层采取 CAS 就不多说了,wc > maximumPoolSize:用来表示如果大于了最大线程数,说明需要减少线程数,至于为什么会出现这种情况,等会理解.(timed && timedOut):说明开启超时退出,且上一次获取任务因超时返回,这个需看后面代码理解.上面的两个条件满足其中之一即可.wc > 1 || workQueue.isEmpty())而这个操作则是为了减少不必要的线程开销,如果阻塞队列为空说明没有任务,那自然不需要多余的线程数去执行,因此会发生接下来的操作,递减线程数,然后跳到下一次循环.


  1. 之后接下来就是从阻塞队列中获取任务的核心了,第一步是根据超时控制字段来决定行为方式,允许超时退出的话,通过poll方式,不允许则通过take方式,两种方式大致是一个等待一定时间,如果为空是前提.另一个是无限等待,会阻塞线程.其具体实现通过阻塞队列的真正实现类别去实现.如果获取到了任务,就返回,如果没有则 timeout 设置为 true,表示没有接受到任务,因此前文的 timeout 就理解了.


  1. 通常而言线程池中的线程数是不允许超过最大线程数的,但通常而言这是一种机制的完整性和规范,假如是自定义线程池的情况下,就有可能出现这种情况,另外一种是本人推测虽然由于增加工作线程数的操作底层是通过 CAS 去实现的,底层是原子性的,同时进行 CAS 操作就有可能导致 ABA 问题出现,或者操作失败,或者不断自旋的可能,


6.任务的提交 submit


众所周知,任务需要进行提交给线程池,再有线程池去执行,而 Runnable 接口实现的 run 方法是没有返回值的,而在线程中 Callable 通常具备返回值,且配备 Future 去接受结果.因此 submit 具备不同的操作


这里以 AbstractExecutorService(线程池的父类)接口为例:


public Future<?> submit(Runnable task) {        if (task == null) throw new NullPointerException();        RunnableFuture<Void> ftask = newTaskFor(task, null);        execute(ftask);        return ftask;    }
 public <T> Future<T> submit(Runnable task, T result) {        if (task == null) throw new NullPointerException();        RunnableFuture<T> ftask = newTaskFor(task, result);        execute(ftask);        return ftask;    }
public <T> Future<T> submit(Callable<T> task) {        if (task == null) throw new NullPointerException();        RunnableFuture<T> ftask = newTaskFor(task);        execute(ftask);        return ftask;    }
复制代码


  • RunnableFuture接口的实现类FutureTask,总而言之就是转换为一个 Runnable,然后进行提交,最后返回一个 future,至于FutureTask具体内容自行详解.

7.线程池的关闭


public void shutdown() {        final ReentrantLock mainLock = this.mainLock;        mainLock.lock();        try {            checkShutdownAccess();            advanceRunState(SHUTDOWN);            interruptIdleWorkers();            onShutdown(); // hook for ScheduledThreadPoolExecutor        } finally {            mainLock.unlock();        }        tryTerminate();    }
复制代码


实现逻辑也很清楚,检查是否可以关闭线程,然后设置线程的状态,interruptIdleWorkers()这个方法算是关键的,他会去中断 worker;onShutdown是一个空方法,留给子类去实现的.


 private void interruptIdleWorkers() {        interruptIdleWorkers(false);    }
复制代码


private void interruptIdleWorkers(boolean onlyOne) {        final ReentrantLock mainLock = this.mainLock;        mainLock.lock();        try {            for (Worker w : workers) {                Thread t = w.thread;                if (!t.isInterrupted() && w.tryLock()) {                    try {                        t.interrupt();                    } catch (SecurityException ignore) {                    } finally {                        w.unlock();                    }                }                if (onlyOne)                    break;            }        } finally {            mainLock.unlock();        }    }
复制代码


他会去遍历集合 workers,获取每一个 worker 的工作线程,然后尝试去中断,最后结束.


文章转载自：不会上猪的树

原文链接：https://www.cnblogs.com/blissful/p/17930818.html

体验地址：http://www.jnpfsoft.com/?from=001

划线
评论
复制
发布于: 刚刚阅读数: 4
Java前端
用户头像

快乐非自愿限量之名

关注

还未添加个人签名 2023-06-19 加入

还未添加个人简介

评论

发布
暂无评论
Copyright © 2023, Geekbang Technology Ltd. All rights reserved. 极客邦控股（北京）有限公司 | 京 ICP 备 16027448 号 - 5京公网安备京公网安备 11010502039052号 | 产品资质
Java线程池ThreadPoolExecutor源码解析_Java_快乐非自愿限量之名_InfoQ写作社区
vConsole