-
线程池源码解析 4.runWorker() 方法
线程池源码解析—runWorker()方法
worker总流程图
runWorker()
- run() 方法调用了 runWorker()，当 worker 创建好并启动后需要执行 run 逻辑获取任务。
大体流程
- 先会执行当前的 worker 内部的任务，执行任务时需要加锁 (worker内部的锁)，如果执行内部任务成功时会尝试去队列中拿任务继续执行（可能会被阻塞）。当执行任务失败时会将当前的 worker 从 workers 集合中删除。
// w 就是创建并启动的worker(Runnable) final void runWorker(Worker w) { // wt 获取到的当前线程 Thread wt = Thread.currentThread(); // 将任务赋值给task Runnable task = w.firstTask; w.firstTask = null; /* * 这里不是解锁操作，这里是为了设置state = 0 以及 ExclusiveOwnerThread = null，因为起始状态state = -1, 不允许任何线程抢占锁， * 这里就是初始化操作。 */ w.unlock(); /* * 表示是否突然退出标志位， * true-> 发生异常了，当前线程突然退出，后面会做处理 * false-> 正常退出 */ boolean completedAbruptly = true; try { /* * 条件1： * task != null 指的就是firstTask是不是null，不是null执行循环体 * * 条件2： * getTask()方法就是当前线程从BlockingQueue中获取任务，此方法会阻塞 * getTask() == null 说明当前线程需要执行结束逻辑 */ while (task != null || (task = getTask()) != null) { /* * 加锁 设置独占锁为当前线程 防止其他线程将当前线程中断 * 为什么要设置独占锁？shutdown时会判断当前worker状态，根据独占锁是否空闲来判断当前worker是否正在工作 */ w.lock(); /* * 这里有两个作用： * 1.线程池处于STOP/TIDYING/TERMINATION状态时需要设置线程的中断标志位 * 2.强制刷新标志位为false，通过Thread.interrupted()方法，因为有可能上一次执行task时，当前线程的中断标志位被设置为了true， * 且没有处理，这里就需要强制刷新一下，不会影响到后面的task。 */ if ((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) && !wt.isInterrupted()) // 设置中断标志位为true wt.interrupt(); try { // 钩子方法，留给子类实现 beforeExecute(wt, task); // 表示异常情况，如果thrown不为空，表示task运行过程中 向上层抛出异常了 Throwable thrown = null; try { // task 可能是FutureTask 也可能是 普通的Runnable接口实现类 // 如果前面是通过submit()提交的 runnable/callable 会被封装成 FutureTask，调用run方法执行业务逻辑 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++; /* * worker处理完一个任务后，会释放掉独占锁 * 1.正常情况下，会再次回到getTask()获取任务(外面是一个while循环) * 2.任务执行出异常 */ w.unlock(); } /* * 什么情况下，回来到这里？ * 当getTask()返回null时，说明当前线程应该执行退出逻辑了 */ completedAbruptly = false; } finally { /* * task.run()内部抛出异常，直接从w.unlock那里跳到这一行 * 正常退出：completedAbruptly = false * 异常退出：completedAbruptly = true */ processWorkerExit(w, completedAbruptly); } }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
- 84
- 85
- 86
- 87
- 88
- 89
- 90
getTask()
- 到当前的任务队列获取任务，如果能获取到就返回，如果获取不到则会阻塞当前线程，当达到某种条件时会返回 null。可能会超时也有可能不会，根据当前线程是否可以被回收决定。
/* * 什么情况下会返回null？ * 1.当前线程状态 >= STOP，成立说明当前的状态最低也是STOP，一定要返回null * 2.线程池状态是SHUTDOWN，并且队列中没有元素 * 3.线程池中的线程数量超过最大限制时，会有一部分线程返回null * 4.线程池中的线程数超过了corePoolSize时，会有一部分线程 获取任务超时后 返回null */ private Runnable getTask() { // 表示当前线程获取任务是否超时，默认是false，true表示已超时 boolean timedOut = false; // 自旋 for (;;) { // 获取最新的ctl的值 int c = ctl.get(); // 获取线程池当前运行状态 int rs = runStateOf(c); /* * 条件1：rs >= SHUTDOWN 条件成立：说明当前线程池是非RUNNING状态，可能是SHUTDOWN/STOP... * 条件2：(rs >= STOP || workQueue.isEmpty()) * 2.1 rs >= STOP 成立说明：当前的状态最低也是STOP状态，一定要返回null了 * 2.2 前置条件：状态是SHUTDOWN(因为 rs >= SHUTDOWN 且 rs >= STOP 不成立，所以当前状态只能为SHUTDOWN) * workQueue.isEmpty()条件成立：说明当前线程池状态为SHUTDOWN状态 且 任务队列已空，此时一定返回null * 返回null，runWorker方法就会将返回null的线程执行线程退出线程池的逻辑。 */ if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { // 使用CAS + 自旋将ctl值-1 decrementWorkerCount(); return null; } /* * 执行到这里，有几种情况？ * 1.线程池是Running状态 * 2.线程池是SHUTDOWN状态，但是队列中还有任务(此时可以创建线程) */ // 获取线程池中的线程数量 int wc = workerCountOf(c); /* * timed == true 表示当前这个线程获取task时是支持超时机制的，使用queue.poll(xx, xx); 第一个参数是时间，第二个参数是时间单位。 * 当获取task超时的情况下，下一次自旋就可能返回null了。 * timed == false 表示当前这个线程 获取 task 时 是不支持超时机制的，当前线程会使用 queue.take(); * 情况1：allowCoreThreadTimeOut == true 表示核心线程数量内的线程 也可以被回收 * 所有线程 都是使用queue.poll(xx, xx) 超时机制这种方式获取task * 情况2：allowCoreThreadTimeOut == false 表示当前线程池会维护核心数量内的线程 * wc > corePoolSize * 条件成立：当前线程池中的线程数量是大于核心线程数的，此时让所有路过这里的线程(救急线程)，都使用poll(xx,xx)支持超时的方式去获取任务 * 这样，就会可能有一部分线程获取不到任务，获取不到任务 返回null，然后runWorker会执行线程退出逻辑。 */ boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; /* * 条件1：(wc > maximumPoolSize || (timed && timedOut)) * 1.1：wc > maximumPoolSize 为什么会成立？setMaximumPoolSize()方法，可能外部线程将线程池最大线程数设置为比初始化时的要小 * 1.2: (timed && timedOut) 条件成立：前置条件，当前线程使用poll方式获取task。上一次循环时 使用poll方式获取任务时，超时了 * 条件1 为true表示 线程可以被回收，达到回收标准，当确实需要回收时再回收。 * 条件2：(wc > 1 || workQueue.isEmpty()) * 2.1: wc > 1 条件成立，说明当前线程池中还有其他线程，当前线程可以直接回收，返回null * 2.2: workQueue.isEmpty() 前置条件 wc == 1， 条件成立：说明当前任务队列 已经空了，最后一个线程，也可以放心的退出。 * 判断当前线程是否达到了回收的标准， * 当获取任务超时并且核心线程可以被回收并且 当前线程池线程数量大于1并且队列中没有任务了，当前线程就达到回收标准了， * 当确实需要回收时才会被回收。 */ if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) { /* * 使用CAS机制 将ctl值-1，减1成功的线程，返回null * CAS有可能会失败，为什么会失败？ * 1.其它线程先你一步退出 * 2.线程池状态发生了变化 */ if (compareAndDecrementWorkerCount(c)) // 直接返回null return null; /* * 如果CAS失败，再次自旋，timed就有可能是false了，因为当前线程CAS失败， * 很有可能是因为其他线程成功退出导致的，再次自旋时检查发现，当前线程就可能不属于回收范围了 */ continue; } try { /* * 获取任务的逻辑 * 根据timed的值，判断去队列中获取任务是使用带超时时间的还是不带超时时间的(注意：如果获取不到任务都会阻塞) */ Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take(); // 获取到了任务 if (r != null) // 返回任务 return r; // 说明当前线程超时了 继续进行自旋 timedOut = true; } catch (InterruptedException retry) { timedOut = false; } } }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
- 84
- 85
- 86
- 87
- 88
- 89
- 90
- 91
- 92
- 93
- 94
- 95
- 96
- 97
processWorkerExit()
- runWorker() 的收尾方法。
/* * @param w：表示当前worker * @param completedAbruptly：true表示当前worker是因为任务出异常退出的 * false表示当前worker是因为没有获取到任务 */ private void processWorkerExit(Worker w, boolean completedAbruptly) { /* * 条件成立：代表当前worker是发生异常退出的，即在执行任务期间向上抛出异常了 * 异常退出时，ctl计数并没有-1 */ if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted // 使用CAS + 自旋将ctl的值-1 decrementWorkerCount(); final ReentrantLock mainLock = this.mainLock; // 加锁(全局锁) mainLock.lock(); try { // 将当前worker完成的任务总数累加到全局总数上 completedTaskCount += w.completedTasks; // 将当前worker从线程池(workers就是一个HashSet)移除 workers.remove(w); } finally { // 解锁 mainLock.unlock(); } // 后面详细讲 tryTerminate(); // 获取最新ctl值 int c = ctl.get(); // 条件成立：当前线程池状态为RUNNING 或者 SHUTDOWN状态 if (runStateLessThan(c, STOP)) { // 这里成立说明当前线程是正常退出 if (!completedAbruptly) { // min表示线程池最低可以持有的线程数量 // allowCoreThreadTimeOut == true => 说明核心线程数内的线程，也会超时被回收 => min == 0 // allowCoreThreadTimeOut == false => min == corePoolSize int min = allowCoreThreadTimeOut ? 0 : corePoolSize; /* * 线程池状态：RUNNING SHUTDOWN * 条件1：假设mid == 0成立 * 条件2：队列不为空 * 说明队列中还有任务，起码得留一个线程 */ if (min == 0 && ! workQueue.isEmpty()) min = 1; // 赋值为1 /* * 条件成立： 线程池中还拥有足够的线程时，后续就不需要调用addWorker()了 * 考虑一个问题： workerCountOf(c) >= min -> (0 >= 0) ？ * 当线程池中的核心线程数是可以被回收的情况下，会出现这种情况，这种情况下， * 当前线程池中的线程数会变为0，下次在提交任务时，会再创建线程。 */ if (workerCountOf(c) >= min) return; } /* * 来到这里的情况 * 1.completedAbruptly为true，表示当前线程在执行task时发生异常退出了，这里一定要创建一个新worker顶上去 * 2.当队列中还有任务时，起码留一个线程，这里就会创建一个线程(worker) * 3.当前线程数 < corePoolSize，此时会创建线程，维护线程池数量在corePoolSize个 */ addWorker(null, false); } }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
参考
- 视频参考
- 文章参考
-
相关阅读:
字节跳动java面试题，附详细答案解析
那些不用写代码也能做游戏的工具
利用经典热门电视剧写爆款公众号爆文10万+阅读量
新手如何快速上手HTTP爬虫IP？
ASCII_Util.java
淘宝分布式文件存储系统( 三 ) -TFS
吴恩达机器学习-可选的实验室-正则化成本和梯度（Regularized Cost and Gradient）
React汇率小案例
计算机图形学实验——2023-09-28
win11安装更新错误0x800f081f怎么解决？
- 原文地址：https://blog.csdn.net/weixin_53407527/article/details/127829844
