文章目录

• 一、线程池中的 Worker ( 工作者 )
• 二、线程池中的工作流程 runWorker
• 三、线程池任务队列中获取任务 getTask

在博客 【Android 异步操作】线程池 ( 线程池 execute 方法源码解析 ) 中 , 讲解 线程池 ThreadPoolExecutor 的 execute 方法时 , 有两个重要的核心方法 ;

两个核心的操作 :

• 添加任务 : addWorker(command, true) , 第二个参数为 true 是添加核心线程任务 , 第二个参数为 false 是添加非核心线程任务 ;
• 拒绝任务 : reject(command)

在上一篇博客 【Android 异步操作】线程池 ( 线程池 reject 拒绝任务 | 线程池 addWorker 添加任务 ) 介绍了 addWorker 添加任务 , reject 拒绝任务 的源码细节 ;

本博客中介绍 Worker ( 工作者 ) 的相关源码

一、线程池中的 Worker ( 工作者 )

---

工作者 Worker 主要 为线程执行任务 , 维护终端控制状态 , 同时记录其它信息 ;

该类扩展了 AbstractQueuedSynchronizer , 目的是 简化 每个任务执行时 获取和释放锁的过程 ;

该操作可以防止中断用于唤醒等待任务的工作线程 , 不会中断一个正在运行的线程 ;

Worker 代码及相关注释说明 :

public class ThreadPoolExecutor extends AbstractExecutorService {/*** 工作者类主要为线程执行任务 , 维护终端控制状态 , 同时记录其它信息 ; * 该类扩展了 AbstractQueuedSynchronizer , 目的是简化 每个任务执行时 获取和释放锁的过程 ; * 该操作可以防止中断用于唤醒等待任务的工作线程 , 不会中断一个正在运行的线程 ; */private final class Workerextends AbstractQueuedSynchronizerimplements Runnable{/*** 该类不会被序列化, 提供该常量用于支持 Java 文档警告 .*/private static final long serialVersionUID = 6138294804551838833L;/** 该工作者运行的线程 , 如果线程工厂创建失败 , 就为空. */final Thread thread;/** 线程初始任务 , 可能为空. */Runnable firstTask;/** 每个线程的任务计数 */volatile long completedTasks;/*** 使用线程工厂 , 根据给定的初始任务 , 创建工作者*/Worker(Runnable firstTask) {setState(-1); // inhibit interrupts until runWorkerthis.firstTask = firstTask;// 线程是在构造函数中 , 使用线程工厂创建的 this.thread = getThreadFactory().newThread(this);}/** 将主要的循环操作委托给了外部的 runWorker , 本博客下面有该方法的解析 . */public void run() {runWorker(this);}// 锁相关方法 //// 0 代表未锁定状态 .// 1 代表锁定状态 .protected boolean isHeldExclusively() {return getState() != 0;}protected boolean tryAcquire(int unused) {if (compareAndSetState(0, 1)) {setExclusiveOwnerThread(Thread.currentThread());return true;}return false;}protected boolean tryRelease(int unused) {setExclusiveOwnerThread(null);setState(0);return true;}public void lock() { acquire(1); }public boolean tryLock() { return tryAcquire(1); }public void unlock() { release(1); }public boolean isLocked() { return isHeldExclusively(); }void interruptIfStarted() {Thread t;if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {try {t.interrupt();} catch (SecurityException ignore) {}}}} }

二、线程池中的工作流程 runWorker

---

/*** Main worker run loop. Repeatedly gets tasks from queue and* executes them, while coping with a number of issues:** 1. We may start out with an initial task, in which case we* don't need to get the first one. Otherwise, as long as pool is* running, we get tasks from getTask. If it returns null then the* worker exits due to changed pool state or configuration* parameters. Other exits result from exception throws in* external code, in which case completedAbruptly holds, which* usually leads processWorkerExit to replace this thread.** 2. Before running any task, the lock is acquired to prevent* other pool interrupts while the task is executing, and then we* ensure that unless pool is stopping, this thread does not have* its interrupt set.** 3. Each task run is preceded by a call to beforeExecute, which* might throw an exception, in which case we cause thread to die* (breaking loop with completedAbruptly true) without processing* the task.** 4. Assuming beforeExecute completes normally, we run the task,* gathering any of its thrown exceptions to send to afterExecute.* We separately handle RuntimeException, Error (both of which the* specs guarantee that we trap) and arbitrary Throwables.* Because we cannot rethrow Throwables within Runnable.run, we* wrap them within Errors on the way out (to the thread's* UncaughtExceptionHandler). Any thrown exception also* conservatively causes thread to die.** 5. After task.run completes, we call afterExecute, which may* also throw an exception, which will also cause thread to* die. According to JLS Sec 14.20, this exception is the one that* will be in effect even if task.run throws.** The net effect of the exception mechanics is that afterExecute* and the thread's UncaughtExceptionHandler have as accurate* information as we can provide about any problems encountered by* user code.** @param w the worker*/final void runWorker(Worker w) {Thread wt = Thread.currentThread();// 拿到了一个任务 Runnable task = w.firstTask;w.firstTask = null;w.unlock(); // allow interruptsboolean completedAbruptly = true;try {// 第一次循环 task 不为空 , 直接就进入了循环体 // 第二次循环 task 为空 , 此时执行 || 后的逻辑 (task = getTask()) != null// 该逻辑中从线程池任务队列中获取任务 , 然后执行该任务 // 此处一直循环读取线程池任务队列中的任务并执行 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 interruptif ((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

---

getTask 从 线程池 任务队列中 获取任务 , 该方法执行 阻塞 或 定时等待 任务 , 具体执行哪个需要根据当前的配置情况 ;

这里通过 线程数 判断该线程是 核心线程 , 还是 非核心线程 ;

非核心线程 :

• 判定条件 : 如果当前执行的线程 大于 核心线程数 , 就是非核心线程
• 获取方法 : 非核心线程 调用 poll 方法从任务队列中取任务
• 线程回收 : 如果超过 keepAliveTime 时间还取不到任务 , 非核心线程 空闲时间 超过了一定时间 , 此时需要回收

核心线程 :

• 获取方法 : 如果该线程是核心线程 , 那么就会调用 take 方法 , 而不是 poll 方法
• 阻塞方法 : take 方法是阻塞的
• 不会被回收 : 核心线程不会回收 , 非核心线程超过一定时间会被回收

如果出现下面 4 中情况 , 工作者必须退出 , 该方法返回 null :

• 工作者数量超过线程池个数
• 线程池停止
• 线程池关闭 , 任务队列清空
• 该工作者等待时间超过空闲时间 , 需要被回收 ; 前提是该线程是非和核心线程 ;

getTask 相关源码 :

/*** 执行阻塞或定时等待任务 , 具体执行哪个需要根据当前的配置情况 ; * * 如果出现下面 4 中情况 , 工作者必须退出 , 该方法返回 null : * 1 . 工作者数量超过线程池个数 * 2 . 线程池停止* 3 . 线程池关闭 , 任务队列清空 * 4 . 该工作者等待时间超过空闲时间 , 需要被回收 ; 前提是该线程是非和核心线程 ; ** @return 返回要执行的任务 ; 如果返回空 , 说明该 工作者 Worker 必须退出 , 工作者计数 -1*/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?// 如果 wc 大于 核心线程数 , 说明本线程是非核心线程 boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;if ((wc > maximumPoolSize || (timed && timedOut))&& (wc > 1 || workQueue.isEmpty())) {if (compareAndDecrementWorkerCount(c))return null;continue;}try {// 这里进行了时间判断 // 如果当前执行的线程 大于 核心线程数 , 就是非核心线程 // 调用 poll 方法从任务队列中取任务, 如果超过 keepAliveTime 时间还取不到任务 , // 非核心线程 空闲时间 超过了一定时间 , 此时需要回收 // 如果该线程是核心线程 , 那么就会调用 take 方法 , 而不是 poll 方法// take 方法是阻塞的 // 因此核心线程不会回收 , 非核心线程超过一定时间会被回收 Runnable r = timed ?workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :workQueue.take();if (r != null)return r;timedOut = true;} catch (InterruptedException retry) {timedOut = false;}}}

总结

以上是生活随笔为你收集整理的【Android 异步操作】线程池 ( Worker 简介 | 线程池中的工作流程 runWorker | 从线程池任务队列中获取任务 getTask )的全部内容，希望文章能够帮你解决所遇到的问题。

如果觉得生活随笔网站内容还不错，欢迎将生活随笔推荐给好友。