文章详情

短信预约-IT技能 免费直播动态提醒

请输入下面的图形验证码

提交验证

短信预约提醒成功

Java线程池源码的深度解析

2022-11-13 18:46

关注

概述

线程池的好处和使用本篇文章就不赘叙了,不了解的可以参考下面两篇文章:

那么本文重点是从源码层面理解jdk8中线程池的实现。

核心机制

再分析源码之前,我们还是先回顾和熟悉下线程的核心工作机制。

线程池工作原理

线程池采用的是一种生产者-消费者的模型,如下图:

线程池状态

线程池的状态有5种,他们的状态转换如上图所示,这里记得区别线程的状态,它们不是一回事。

ThreadPoolExecutor类存放线程池的状态信息很特别,是存储在一个int类型原子变量的高3位,而低29位用来存储线程池当前运行的线程数量。通过将线程池的状态和线程数量合二为一,可以做到一次CAS原子操作更新数据。

状态高3位值说明
RUNNING111运行状态,线程池被创建后的初始状态,能接受新提交的任务,也能处理阻塞队列中的任务。
SHUTDOWN000关闭状态,不再接受新提交的任务,但任可以处理阻塞队列中的任务。
STOP001停止状态,会中断正在处理的线程,不能接受新提交的任务,也不会处理阻塞队列中的任务。
TIDYING010所有任务都已经终止,有效工作线程为0。
TERMINATED011终止状态,线程池彻底终止。

源码解析

上图是线程池核心类ThreadPoolExecutor的类结构图:

关键成员变量

1.线程池的状态信息和线程数量信息(ctl)相关

线程的状态信息和数量信息用同一个int的原子变量存储,高3位存储状态信息,低29位存储线程数量。

// ctl,原子变量,存储状态和线程数量,初始化运行状态+0
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
// 静态常量,表示线程数量存放的位数29=32-3
private static final int COUNT_BITS = Integer.SIZE - 3;
// 线程数量最大的容量,低 COUNT_BITS 位所能表达的最大数值,000 11111111111111111111 => 5亿多
private static final int CAPACITY  = (1 << COUNT_BITS) - 1;

通过位运算符设置各个状态的高三位值。

// 111 000000000000000000,转换成整数后其实就是一个【负数】
private static final int RUNNING    = -1 << COUNT_BITS;
// 000 000000000000000000
private static final int SHUTDOWN   =  0 << COUNT_BITS;
// 001 000000000000000000
private static final int STOP       =  1 << COUNT_BITS;
// 010 000000000000000000
private static final int TIDYING    =  2 << COUNT_BITS;
// 011 000000000000000000
private static final int TERMINATED =  3 << COUNT_BITS;

从ctl中获取线程池的状态值

// ~CAPACITY = ~000 11111111111111111111 = 111 000000000000000000000(取反)
// &运算符,和1&是它本身,和0&就是0,就可以获得高位值。
private static int runStateOf(int c)     { return c & ~CAPACITY; }

从ctl中获取线程池的数量

// CAPACITY = 000 11111111111111111111
// &运算符,和1&是它本身,和0&就是0,就可以获得低29位
private static int workerCountOf(int c)  { return c & CAPACITY; }

生成ctl值

// rs 表示线程池状态,wc 表示当前线程池中 worker(线程)数量,相与以后就是合并后的状态
private static int ctlOf(int rs, int wc) { return rs | wc; }

比较当前线程池 ctl 所表示的状态

线程池状态值的大小关系:RUNNING < SHUTDOWN < STOP < TIDYING < TERMINATED

// 比较当前线程池 ctl 所表示的状态,是否小于某个状态 s
private static boolean runStateLessThan(int c, int s) { return c < s; }
// 比较当前线程池 ctl 所表示的状态,是否大于等于某个状态s
private static boolean runStateAtLeast(int c, int s) { return c >= s; }
// 小于 SHUTDOWN 的一定是 RUNNING,SHUTDOWN == 0
private static boolean isRunning(int c) { return c < SHUTDOWN; }

cas设置ctl的值

// 使用 CAS 方式 让 ctl 值 +1 ,成功返回 true, 失败返回 false
private boolean compareAndIncrementWorkerCount(int expect) {
    return ctl.compareAndSet(expect, expect + 1);
}
// 使用 CAS 方式 让 ctl 值 -1 ,成功返回 true, 失败返回 false
private boolean compareAndDecrementWorkerCount(int expect) {
    return ctl.compareAndSet(expect, expect - 1);
}
// 将 ctl 值减一,do while 循环会一直重试,直到成功为止
private void decrementWorkerCount() {
    do {} while (!compareAndDecrementWorkerCount(ctl.get()));
}

2.线程池中的队列

// 线程池用于保存任务并将任务传递给工作线程的队列
private final BlockingQueue<Runnable> workQueue;

3.控制并发的锁

// 增加减少 worker 或者时修改线程池运行状态需要持有 mainLock
private final ReentrantLock mainLock = new ReentrantLock();

4.线程池中工作线程的集合

private final HashSet<Worker> workers = new HashSet<Worker>();

5.线程池构造参数关系属性

// 核心线程数量
private volatile int corePoolSize;
// 线程池最大线程数量
private volatile int maximumPoolSize;	
// 空闲线程存活时间
private volatile long keepAliveTime;	
// 创建线程时使用的线程工厂,默认是 DefaultThreadFactory
private volatile ThreadFactory threadFactory;	
// 【超过核心线程提交任务就放入 阻塞队列】
private final BlockingQueue<Runnable> workQueue;
// 拒绝策略
private volatile RejectedExecutionHandler handler;	

6.线程池监控相关属性

// 记录线程池生命周期内线程数最大值
private int largestPoolSize;	
// 记录线程池所完成任务总数,当某个 worker 退出时将完成的任务累加到该属性
private long completedTaskCount;	

线程提交原理

线程池提交线程有多种方式如execute、submit或者invoke相关方法,我们重点关注在最基础的execute()方法提交任务,把它搞清楚了,其他的都不在话下。

execute(Runnable command)方法是线程提交的入口方法。

//  ThreadPoolExecutor#execute
public void execute(Runnable command) {
        // 如果任务为空,直接抛空指针
        if (command == null)
            throw new NullPointerException();
        // 获取ctl的值,其中高3位是状态信息,低3位是线程数量
        int c = ctl.get();
        // workerCountOf获取当前线程的数量
        // 当前线程数量小于核心线程数,调用addWorker创建一个工作线程
        if (workerCountOf(c) < corePoolSize) {
            // 调用addWorker方法创建工作线程,直接执行任务。如果成功的话,直接结束方法。
            if (addWorker(command, true))
                return;
            // 由于并发等原因,addWorker添加失败,会走到这里,再次获取ctl的值
            c = ctl.get();
        }
    	// 如果线程池是运行状态的话,就把任务加入到队列中
        if (isRunning(c) && workQueue.offer(command)) {
            // 双重检查,因为从上次检查到进入此方法,线程池可能已成为SHUTDOWN状态
            int recheck = ctl.get();
            // 如果发现线程池不是运行状态的话,那就移除这个任务
            if (!isRunning(recheck) && remove(command))
                // 任务出队成功,走拒绝策略
                reject(command);
             // 执行到这说明线程池是 running 状态,获取线程池中的线程数量,判断是否是 0
             // 【担保机制】,保证线程池在 running 状态下,最起码得有一个线程在工作
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        // 走到这里说明线程不是运行状态,或者就是队列满了,offer返回false
        // 再次调用addWoker创建新的线程,如果不成功(一般是超过了线程池最大线程数量),执行拒绝策略
        else if (!addWorker(command, false))
            // 执行拒绝策略
            reject(command);
    }

这个方法是提交线程的主干逻辑:

addWorker方法也是一个很关键的方法, 添加线程到线程池,返回 true 表示创建 Worker 成功,且启动线程。

//  ThreadPoolExecutor#addWorker
// core == true 表示采用核心线程数量限制,false 表示采用 maximumPoolSize
private boolean addWorker(Runnable firstTask, boolean core) {
     // 自旋【判断当前线程池状态是否允许创建线程】,允许就设置线程数量 + 1
    retry:
    for (;;) {
         // 获取 ctl 的值
        int c = ctl.get();
        // 获取当前线程池运行状态
        int rs = runStateOf(c);

         // 判断当前线程池状态【是否允许添加线程】
        
       // 如果线程池状态大于SHUTDOWN 或者是SHUTDOWN状态,队列是空了的话,都不允许创建新的线程
        if (rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN &&
               firstTask == null &&
               ! workQueue.isEmpty()))
            // false,没有创建线程
            return false;
        // 再次自旋
        for (;;) {
            // 获取线程池中线程数量
            int wc = workerCountOf(c);
            // 如果线程数量超过阈值的话,返回false
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
          // 记录线程数量已经加 1,类比于申请到了一块令牌,条件失败说明其他线程修改了数量
            if (compareAndIncrementWorkerCount(c))
                // 申请成功,跳出了 retry 这个 for 自旋
                break retry;
             // CAS 失败,没有成功的申请到令牌
            c = ctl.get(); 
            // 判断当前线程池状态是否发生过变化,被其他线程修改了,可能其他线程调用了 shutdown() 方法
            if (runStateOf(c) != rs)
                // 重新回到retry的执行点
                continue retry;
            // else CAS failed due to workerCount change; retry inner loop
        }
    }

    // 下面开始真正创建线程了
    // 运行标记,表示创建的 worker 是否已经启动,false未启动  true启动
    boolean workerStarted = false;
    // 添加标记,表示创建的 worker 是否添加到池子中了,默认false未添加,true是添加。
    boolean workerAdded = false;
    Worker w = null;
    try {
        //【创建 Worker,底层通过线程工厂 newThread 方法创建执行线程,指定了首先执行的任务】
        w = new Worker(firstTask);
        // 将新创建的 worker 节点中的线程赋值给 t
        final Thread t = w.thread;
        // 这里的判断为了防止 程序员自定义的 ThreadFactory 实现类有 bug,创造不出线程
        if (t != null) {
            final ReentrantLock mainLock = this.mainLock;
            // 加互斥锁,要添加 worker 了
            mainLock.lock();
            try {
                 // 获取最新线程池运行状态
                int rs = runStateOf(ctl.get());
            	// 判断线程池是否为RUNNING状态,不是再【判断当前是否为SHUTDOWN状态且firstTask为空,特殊情况】
                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) {
                     // 当线程start后,线程isAlive会返回true,这里还没开始启动线程,如果被启动了就需要报错
                    if (t.isAlive()) 
                        throw new IllegalThreadStateException();
                    //将新建的 Worker 添加到线程池中
                    workers.add(w);
                    int s = workers.size();
                    // 当前池中的线程数量是一个新高,更新 largestPoolSize
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                     // 添加标记置为 true
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
             // 添加成功就【启动线程执行任务】
            if (workerAdded) {
                // 启动线程
                t.start();
                // 运行标记置为 true
                workerStarted = true;
            }
        }
    } finally {
        // 线程启动失败
        if (! workerStarted)
            // 清理工作,比如从线程池中移除。
            addWorkerFailed(w);
    }
    return workerStarted;
}

private void addWorkerFailed(Worker w) {
    final ReentrantLock mainLock = this.mainLock;
    // 持有线程池全局锁,因为操作的是线程池相关的东西
    mainLock.lock();
    try {
        //条件成立需要将 worker 在 workers 中清理出去。
        if (w != null)
            workers.remove(w);
        // 将线程池计数 -1,相当于归还令牌。
        decrementWorkerCount();
        // 尝试停止线程池
        tryTerminate();
    } finally {
        //释放线程池全局锁。
        mainLock.unlock();
    }
}

这里注意一个点,SHUTDOWN 状态也能添加线程,但是要求新加的 Woker 没有 firstTask,而且当前 queue 不为空,所以创建一个线程来帮助线程池执行队列中的任务。

Woker运行原理

Woker类是ThreadPoolExecutor类的内部类,见明知意,它是承担了一个“工人”干活,也就是工作线程的责任。

1.Worker类

每个 Worker 对象有一个初始任务,启动 Worker 时优先执行,这也是造成线程池不公平的原因。Worker 继承自 AQS,本身具有锁的特性,采用独占锁模式,state = 0 表示未被占用,> 0 表示被占用,< 0 表示初始状态不能被抢锁。

private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
	// worker 内部封装的工作线程
    final Thread thread;		
    // worker 第一个执行的任务,普通的 Runnable 实现类或者是 FutureTask
    Runnable firstTask;	
    // 记录当前 worker 所完成任务数量
    volatile long completedTasks;	
    
    // 构造方法
    Worker(Runnable firstTask) {
        // 设置AQS独占模式为初始化中状态,这个状态不能被抢占锁
       	setState(-1);
        // firstTask不为空时,当worker启动后,内部线程会优先执行firstTask,执行完后会到queue中去获取下个任务
        this.firstTask = firstTask;
        // 使用线程工厂创建一个线程,并且【将当前worker指定为Runnable】,所以thread启动时会调用 worker.run()
        this.thread = getThreadFactory().newThread(this);
    }
    // 不可重入锁,重写了AQS中的方法
    protected boolean tryAcquire(int unused) {
        if (compareAndSetState(0, 1)) {
            setExclusiveOwnerThread(Thread.currentThread());
            return true;
        }
        return false;
    }

 protected boolean tryRelease(int unused) {
        setExclusiveOwnerThread(null);
        // 设置state为0,开始抢锁
        setState(0);
        return true;
    }
}

2.Worker的工作方法run

// Worker#run
public void run() {
    // 调用自身的runWoker方法
    runWorker(this);
}
// Worker#runWorker
final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    // 获取 worker 的 firstTask
    Runnable task = w.firstTask;
    // 引用置空,【防止复用该线程时重复执行该任务】
    w.firstTask = null;
	// 初始化 worker 时设置 state = -1,表示不允许抢占锁
    // 这里需要设置 state = 0 和 exclusiveOwnerThread = null,开始独占模式抢锁
    w.unlock(); 
   // true 表示发生异常退出,false 表示正常退出。
    boolean completedAbruptly = true;
    try {
        // firstTask 不是 null 就直接运行,否则去 queue 中获取任务
        while (task != null || (task = getTask()) != null) {
            // worker 加锁,shutdown 时会判断当前 worker 状态,【根据独占锁状态判断是否空闲】
            w.lock();
            // 说明线程池状态大于 STOP,目前处于 STOP/TIDYING/TERMINATION,此时给线程一个中断信号
            if ((runStateAtLeast(ctl.get(), STOP) ||
                 (Thread.interrupted() &&
                  runStateAtLeast(ctl.get(), STOP))) &&
                // 线程不是处于中断的情况
                !wt.isInterrupted())
                 // 中断线程,设置线程的中断标志位为 true
                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,代表任务执行完成
                task = null;
                // 更新worker完成任务数量
                w.completedTasks++;
                // 解锁
                w.unlock();
            }
        }
         // getTask()方法返回null时会走到这里,表示queue为空并且线程空闲超过保活时间,【当前线程执行退出逻辑】
        completedAbruptly = false;
    } finally {
        // 正常退出 completedAbruptly = false
       	// 异常退出 completedAbruptly = true,【从 task.run() 内部抛出异常】时,跳到这一行
        processWorkerExit(w, completedAbruptly);
    }
}

3.getTask() 获取任务

这个方法主要做了下面几件事情:

private Runnable getTask() {
     // 超时标记,表示当前线程获取任务是否超时,true 表示已超时
    boolean timedOut = false; 

    for (;;) {
        int c = ctl.get();
         // 获取线程池当前运行状态
        int rs = runStateOf(c);

        // 如果发现线程池被关闭了,直接返回null
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            // 使用 CAS 自旋的方式让 ctl 值 -1
            decrementWorkerCount();
            return null;
        }

        // 获取线程池中的线程数量
        int wc = workerCountOf(c);

        //timed用来判断当前线程是否超过一定时间没有获取任务就进行销毁回收,true是需要,false不需要, 有两种情况
        //1. allowCoreThreadTimeOut为true代表允许回收核心线程,那就无所谓了,全部线程都执行超时回收
        //2. 线程数量大于核心线程数,当前线程认为是非核心线程
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

        // 同时满足下面1和2条件下,说明线程要回收,直接返回null
        // 1. 如果线程数量超过最大线程数 或者 上面的timed和超时时间timedOut都为true
        if ((wc > maximumPoolSize || (timed && timedOut))
            // 2.如果线程数量大于1并且队列时空的情况
            && (wc > 1 || workQueue.isEmpty())) {
            // 使用 CAS 机制将 ctl 值 -1 ,减 1 成功的线程,返回 null,代表可以退出
            if (compareAndDecrementWorkerCount(c))
                return null;
            continue;
        }

        try {
            // 从队列中获取任务,有下面两种方法
            // timed为true, 调用超时方法poll获取任务
            // timed为false,调用阻塞方法take获取
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
            if (r != null)
                return r;
            获取任务为 null 说明超时了,将超时标记设置为 true,进入下一次循环,就可以销毁这个线程了
            timedOut = true;
        } catch (InterruptedException retry) {
             // 阻塞线程被打断后超时标记置为 false,【说明被打断不算超时】,要继续获取,直到超时或者获取到任务
            // 如果线程池 SHUTDOWN 状态下的打断,会在循环获取任务前判断,返回 null
            timedOut = false;
        }
    }
}

4.processWorkerExit()工作线程退出方法

// 正常退出 completedAbruptly = false,异常退出为 true
private void processWorkerExit(Worker w, boolean completedAbruptly) {
    // 条件成立代表当前 worker 是发生异常退出的,task 任务执行过程中向上抛出异常了
    if (completedAbruptly) 
        // 从异常时到这里 ctl 一直没有 -1,需要在这里 -1
        decrementWorkerCount();

    final ReentrantLock mainLock = this.mainLock;
    // 加锁
    mainLock.lock();
    try {
        // 将当前 worker 完成的 task 数量,汇总到线程池的 completedTaskCount
        completedTaskCount += w.completedTasks;
		// 将 worker 从线程池中移除
        workers.remove(w);
    } finally {
        mainLock.unlock();	// 解锁
    }
	// 尝试停止线程池,唤醒下一个线程
    tryTerminate();

    int c = ctl.get();
    // 线程池不是停止状态就应该有线程运行【担保机制】
    if (runStateLessThan(c, STOP)) {
        // 正常退出的逻辑,是对空闲线程回收,不是执行出错
        if (!completedAbruptly) {
            // 根据是否回收核心线程确定【线程池中的线程数量最小值】
            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
            // 最小值为 0,但是线程队列不为空,需要一个线程来完成任务担保机制
            if (min == 0 && !workQueue.isEmpty())
                min = 1;
            // 线程池中的线程数量大于最小值可以直接返回
            if (workerCountOf(c) >= min)
                return;
        }
        // 执行 task 时发生异常,有个线程因为异常终止了,需要添加
        // 或者线程池中的数量小于最小值,这里要创建一个新 worker 加进线程池
        addWorker(null, false);
    }
}

总结

本文主要从源码层面分析了线程池的运行机理,总算知道了execute方法背后是如何运转的。

以上就是Java线程池源码的深度解析的详细内容,更多关于Java线程池的资料请关注编程网其它相关文章!

阅读原文内容投诉

免责声明:

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析(完整版)

    难度     813人已做
    查看
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     354人已做
    查看
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     318人已做
    查看
  • 2024年上半年软考高项第一、二批次真题考点汇总(完整版)

    难度     435人已做
    查看
  • 2024年上半年系统架构设计师考试综合知识真题

    难度     224人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机
位置:首页-资讯-后端开发
咦!没有更多了?去看看其它编程学习网 内容吧
首页课程
资料下载
问答资讯