Android Dispatchers.IO
LeeDuo. 人气:0一. Dispatchers.IO
1.Dispatchers.IO
在协程中，当需要执行IO任务时，会在上下文中指定Dispatchers.IO来进行线程的切换调度。 而IO实际上是CoroutineDispatcher类型的对象，实际的值为DefaultScheduler类的常量对象IO，代码如下：
public actual object Dispatchers {
...
@JvmStatic
public val IO: CoroutineDispatcher = DefaultScheduler.IO
}
2.DefaultScheduler类
DefaultScheduler类继承自ExperimentalCoroutineDispatcher类，内部提供了类型为LimitingDispatcher的IO对象，代码如下：
// 系统配置变量
public const val IO_PARALLELISM_PROPERTY_NAME: String = "kotlinx.coroutines.io.parallelism"
...
// 表示不会阻塞的任务，纯CPU任务
internal const val TASK_NON_BLOCKING = 0
// 表示执行过程中可能会阻塞的任务，非纯CPU任务
internal const val TASK_PROBABLY_BLOCKING = 1
...
// 默认线程池名称
internal const val DEFAULT_DISPATCHER_NAME = "Dispatchers.Default"
...
internal object DefaultScheduler : ExperimentalCoroutineDispatcher() {
// 创建名为Dispatchers.IO的线程池
// 最大并发数量为kotlinx.coroutines.io.parallelism指定的值，默认为64与CPU数量中的较大者
// 默认的执行的任务类型为TASK_PROBABLY_BLOCKING
val IO: CoroutineDispatcher = LimitingDispatcher(
this,
systemProp(IO_PARALLELISM_PROPERTY_NAME, 64.coerceAtLeast(AVAILABLE_PROCESSORS)),
"Dispatchers.IO",
TASK_PROBABLY_BLOCKING
)
override fun close() {
throw UnsupportedOperationException("$DEFAULT_DISPATCHER_NAME cannot be closed")
}
// 可以看出IO和Default共用一个线程池
override fun toString(): String = DEFAULT_DISPATCHER_NAME
@InternalCoroutinesApi
@Suppress("UNUSED")
public fun toDebugString(): String = super.toString()
}3.LimitingDispatcher类
LimitingDispatcher类继承自ExecutorCoroutineDispatcher类，实现了TaskContext接口和Executor接口。
LimitingDispatcher类的核心是构造方法中类型为ExperimentalCoroutineDispatcher的dispatcher对象。
LimitingDispatcher类看起来是一个标准的线程池，但实际上LimitingDispatcher类只对类参数中传入的dispatcher进行包装和功能扩展。如同名字中的litmit一样，LimitingDispatcher类主要用于对任务执行数量进行限制，代码如下：
// dispatcher参数传入了DefaultScheduler对象
// parallelism表示并发执行的任务数量
// name表示线程池的名字
// taskMode表示任务模式，TaskContext接口中的常量
private class LimitingDispatcher(
private val dispatcher: ExperimentalCoroutineDispatcher,
private val parallelism: Int,
private val name: String?,
override val taskMode: Int
) : ExecutorCoroutineDispatcher(), TaskContext, Executor {
// 用于保存任务的队列
private val queue = ConcurrentLinkedQueue<Runnable>()
// 用于记录当前正在执行的任务的数量
private val inFlightTasks = atomic(0)
// 获取当前线程池
override val executor: Executor
get() = this
// Executor接口的实现，线程池的核心方法，通过dispatch实现
override fun execute(command: Runnable) = dispatch(command, false)
override fun close(): Unit = error("Close cannot be invoked on LimitingBlockingDispatcher")
// CoroutineDispatcher接口的实现
override fun dispatch(context: CoroutineContext, block: Runnable) = dispatch(block, false)
// 任务分发的核心方法
private fun dispatch(block: Runnable, tailDispatch: Boolean) {
// 获取当前要执行的任务
var taskToSchedule = block
// 死循环
while (true) {
// 当前执行的任务数加一，也可理解生成生成当前要执行的任务的编号
val inFlight = inFlightTasks.incrementAndGet()
// 如果当前需要执行的任务数小于允许的并发执行任务数量，说明可以执行，
if (inFlight <= parallelism) {
// 调用参数中的dispatcher对象，执行任务
dispatcher.dispatchWithContext(taskToSchedule, this, tailDispatch)
// 返回，退出循环
return
}
// 如果达到的最大并发数的限制，则将任务加入到队列中
queue.add(taskToSchedule)
// 下面的代码防止线程竞争导致任务卡在队列里不被执行，case如下：
// 线程1:inFlightTasks = 1 ，执行任务
// 线程2:inFlightTasks = 2，当前达到了parallelism限制，
// 线程1:执行结束，inFlightTasks = 1
// 线程2:将任务添加到队列里，执行结束，inFlightTasks = 0
// 由于未执行，因此这里当前执行的任务数先减一
// 减一后如果仍然大于等于在大并发数，则直接返回，退出循环
if (inFlightTasks.decrementAndGet() >= parallelism) {
return
}
// 如果减一后，发现可以执行任务，则从队首获取任务，进行下一次循环
// 如果队列为空，说明没有任务，则返回，退出循环
taskToSchedule = queue.poll() ?: return
}
}
// CoroutineDispatcher接口的实现，用于yield挂起协程时的调度处理
override fun dispatchYield(context: CoroutineContext, block: Runnable) {
// 也是通过dispatch方法实现，注意这里tailDispatch参数为true
dispatch(block, tailDispatch = true)
}
override fun toString(): String {
return name ?: "${super.toString()}[dispatcher = $dispatcher]"
}
// TaskContext接口的实现，用于在一个任务执行完进行回调
override fun afterTask() {
// 从队首获取一个任务
var next = queue.poll()
// 若可以获取到
if (next != null) {
// 则执行任务，注意这里tailDispatch参数为true
dispatcher.dispatchWithContext(next, this, true)
// 返回
return
}
// 任务执行完毕，当前执行的任务数量减一
inFlightTasks.decrementAndGet()
// 下面的代码防止线程竞争导致任务卡在队列里不被执行，case如下：
// 线程1:inFlightTasks = 1 ，执行任务
// 线程2:inFlightTasks = 2
// 线程1:执行结束，执行afterTask方法，发现队列为空，此时inFlightTasks = 2
// 线程2:inFlightTasks当前达到了parallelism限制，
// 将任务加入到队列中，执行结束，inFlightTasks = 1
// 线程1:inFlightTasks=1，执行结束
// 从队列中取出任务，队列为空则返回
next = queue.poll() ?: return
// 执行任务，注意这里tailDispatch参数为true
dispatch(next, true)
}
}dispatcher的dispatch方法定义在ExperimentalCoroutineDispatcher类中。
4.ExperimentalCoroutineDispatcher类
ExperimentalCoroutineDispatcher类继承自ExecutorCoroutineDispatcher类，代码如下：
// corePoolSize线程池核心线程数
// maxPoolSize表示线程池最大线程数
// schedulerName表示内部协程调度器的名字
// idleWorkerKeepAliveNs表示空闲的线程存活时间
@InternalCoroutinesApi
public open class ExperimentalCoroutineDispatcher(
private val corePoolSize: Int,
private val maxPoolSize: Int,
private val idleWorkerKeepAliveNs: Long,
private val schedulerName: String = "CoroutineScheduler"
) : ExecutorCoroutineDispatcher() {
// 我们在DefaultScheduler类中就是通过默认的构造方法，
// 创建的父类ExperimentalCoroutineDispatcher对象
public constructor(
corePoolSize: Int = CORE_POOL_SIZE,
maxPoolSize: Int = MAX_POOL_SIZE,
schedulerName: String = DEFAULT_SCHEDULER_NAME
) : this(corePoolSize, maxPoolSize, IDLE_WORKER_KEEP_ALIVE_NS, schedulerName)
...
// 创建coroutineScheduler对象
private var coroutineScheduler = createScheduler()
// 核心的分发方法
override fun dispatch(context: CoroutineContext, block: Runnable): Unit =
try {
// 调用coroutineScheduler对象的dispatch方法
coroutineScheduler.dispatch(block)
} catch (e: RejectedExecutionException) {
// 只有当coroutineScheduler正在关闭时，才会拒绝执行，抛出异常
DefaultExecutor.dispatch(context, block)
}
...
private fun createScheduler() = CoroutineScheduler(corePoolSize, maxPoolSize, idleWorkerKeepAliveNs, schedulerName)
...
}
// 核心线程数
@JvmField
internal val CORE_POOL_SIZE = systemProp(
"kotlinx.coroutines.scheduler.core.pool.size",
AVAILABLE_PROCESSORS.coerceAtLeast(2), // !!! at least two here
minValue = CoroutineScheduler.MIN_SUPPORTED_POOL_SIZE
)
// 最大线程数
@JvmField
internal val MAX_POOL_SIZE = systemProp(
"kotlinx.coroutines.scheduler.max.pool.size",
(AVAILABLE_PROCESSORS * 128).coerceIn(
CORE_POOL_SIZE,
CoroutineScheduler.MAX_SUPPORTED_POOL_SIZE
),
maxValue = CoroutineScheduler.MAX_SUPPORTED_POOL_SIZE
)
// 空闲线程的存活时间
@JvmField
internal val IDLE_WORKER_KEEP_ALIVE_NS = TimeUnit.SECONDS.toNanos(
systemProp("kotlinx.coroutines.scheduler.keep.alive.sec", 60L)
)在ExperimentalCoroutineDispatcher类的dispatch方法内部，通过调用类型为CoroutineScheduler的对象的dispatch方法实现。
二.CoroutineScheduler类
1.CoroutineScheduler类的继承关系
在对CoroutineScheduler类的dispatch方法分析之前，首先分析一下CoroutineScheduler类的继承关系，代码如下：
// 实现了Executor和Closeable接口
// corePoolSize线程池核心线程数
// maxPoolSize表示线程池最大线程数
// schedulerName表示内部协程调度器的名字
// idleWorkerKeepAliveNs表示空闲的线程存活时间
internal class CoroutineScheduler(
@JvmField val corePoolSize: Int,
@JvmField val maxPoolSize: Int,
@JvmField val idleWorkerKeepAliveNs: Long = IDLE_WORKER_KEEP_ALIVE_NS,
@JvmField val schedulerName: String = DEFAULT_SCHEDULER_NAME
) : Executor, Closeable {
init {
// 核心线程数量必须大于等于MIN_SUPPORTED_POOL_SIZE
require(corePoolSize >= MIN_SUPPORTED_POOL_SIZE) {
"Core pool size $corePoolSize should be at least $MIN_SUPPORTED_POOL_SIZE"
}
// 最大线程数量必须大于等于核心线程数量
require(maxPoolSize >= corePoolSize) {
"Max pool size $maxPoolSize should be greater than or equals to core pool size $corePoolSize"
}
// 最大线程数量必须小于等于MAX_SUPPORTED_POOL_SIZE
require(maxPoolSize <= MAX_SUPPORTED_POOL_SIZE) {
"Max pool size $maxPoolSize should not exceed maximal supported number of threads $MAX_SUPPORTED_POOL_SIZE"
}
// 空闲的线程存活时间必须大于0
require(idleWorkerKeepAliveNs > 0) {
"Idle worker keep alive time $idleWorkerKeepAliveNs must be positive"
}
}
...
// Executor接口中的实现，通过dispatch方法实现
override fun execute(command: Runnable) = dispatch(command)
// Closeable接口中的实现，通过shutdown方法实现
override fun close() = shutdown(10_000L)
...
}2.CoroutineScheduler类的全局变量
接下来对CoroutineScheduler类中重要的全局变量进行分析，代码如下：
// 用于存储全局的纯CPU(不阻塞)任务
@JvmField
val globalCpuQueue = GlobalQueue()
// 用于存储全局的执行非纯CPU(可能阻塞)任务
@JvmField
val globalBlockingQueue = GlobalQueue()
...
// 用于记录当前处于Parked状态(一段时间后自动终止)的线程的数量
private val parkedWorkersStack = atomic(0L)
...
// 用于保存当前线程池中的线程
// workers[0]永远为null，作为哨兵位
// index从1到maxPoolSize为有效线程
@JvmField
val workers = AtomicReferenceArray<Worker?>(maxPoolSize + 1)
...
// 控制状态
private val controlState = atomic(corePoolSize.toLong() shl CPU_PERMITS_SHIFT)
// 表示已经创建的线程的数量
private val createdWorkers: Int inline get() = (controlState.value and CREATED_MASK).toInt()
// 表示可以获取的CPU令牌数量，初始值为线程池核心线程数量
private val availableCpuPermits: Int inline get() = availableCpuPermits(controlState.value)
// 获取指定的状态的已经创建的线程的数量
private inline fun createdWorkers(state: Long): Int = (state and CREATED_MASK).toInt()
// 获取指定的状态的执行阻塞任务的数量
private inline fun blockingTasks(state: Long): Int = (state and BLOCKING_MASK shr BLOCKING_SHIFT).toInt()
// 获取指定的状态的CPU令牌数量
public inline fun availableCpuPermits(state: Long): Int = (state and CPU_PERMITS_MASK shr CPU_PERMITS_SHIFT).toInt()
// 当前已经创建的线程数量加1
private inline fun incrementCreatedWorkers(): Int = createdWorkers(controlState.incrementAndGet())
// 当前已经创建的线程数量减1
private inline fun decrementCreatedWorkers(): Int = createdWorkers(controlState.getAndDecrement())
// 当前执行阻塞任务的线程数量加1
private inline fun incrementBlockingTasks() = controlState.addAndGet(1L shl BLOCKING_SHIFT)
// 当前执行阻塞任务的线程数量减1
private inline fun decrementBlockingTasks() {
controlState.addAndGet(-(1L shl BLOCKING_SHIFT))
}
// 尝试获取CPU令牌
private inline fun tryAcquireCpuPermit(): Boolean = controlState.loop { state ->
val available = availableCpuPermits(state)
if (available == 0) return false
val update = state - (1L shl CPU_PERMITS_SHIFT)
if (controlState.compareAndSet(state, update)) return true
}
// 释放CPU令牌
private inline fun releaseCpuPermit() = controlState.addAndGet(1L shl CPU_PERMITS_SHIFT)
// 表示当前线程池是否关闭
private val _isTerminated = atomic(false)
val isTerminated: Boolean get() = _isTerminated.value
companion object {
// 用于标记一个线程是否在parkedWorkersStack中(处于Parked状态)
@JvmField
val NOT_IN_STACK = Symbol("NOT_IN_STACK")
// 线程的三个状态
// CLAIMED表示线程可以执行任务
// PARKED表示线程暂停执行任务，一段时间后会自动进入终止状态
// TERMINATED表示线程处于终止状态
private const val PARKED = -1
private const val CLAIMED = 0
private const val TERMINATED = 1
// 以下五个常量为掩码
private const val BLOCKING_SHIFT = 21 // 2x1024x1024
// 1-21位
private const val CREATED_MASK: Long = (1L shl BLOCKING_SHIFT) - 1
// 22-42位
private const val BLOCKING_MASK: Long = CREATED_MASK shl BLOCKING_SHIFT
// 42
private const val CPU_PERMITS_SHIFT = BLOCKING_SHIFT * 2
// 43-63位
private const val CPU_PERMITS_MASK = CREATED_MASK shl CPU_PERMITS_SHIFT
// 以下两个常量用于require中参数判断
internal const val MIN_SUPPORTED_POOL_SIZE = 1
// 2x1024x1024-2
internal const val MAX_SUPPORTED_POOL_SIZE = (1 shl BLOCKING_SHIFT) - 2
// parkedWorkersStack的掩码
private const val PARKED_INDEX_MASK = CREATED_MASK
// inv表示01反转
private const val PARKED_VERSION_MASK = CREATED_MASK.inv()
private const val PARKED_VERSION_INC = 1L shl BLOCKING_SHIFT
}CoroutineScheduler类中对线程的状态与权限控制：
availableCpuPermits的初始值为参数中核心线程数corePoolSize的值，表示CoroutineScheduler类中最多只有corePoolSize个核心线程。执行纯CPU任务的线程每次执行任务之前需要在availableCpuPermits中进行记录与申请。blockingTasks表示执行非纯CPU任务的数量。这部分线程在执行时不需要CPU令牌。createdWorkers表示当前线程池中所有线程的数量，每个线程在创建或终止时都需要通过在这里进行记录。这些变量的具体关系如下：
createdWorkers = blockingTasks + corePoolSize - availableCpuPermits
CPU令牌是线程池自定义的概念，不代表时间片，只是为了保证核心线程的数量。
三.Worker类与WorkerState类
在分析CoroutineScheduler类的dispatch方法之前，还需要分析一下CoroutineScheduler类中的两个重要的内部类Worker类以及其对应的状态类WorkerState类。
Worker是一个线程池中任务的核心执行者，几乎在所有的线程池中都存在Worker的概念。
1.WorkerState类
首先分析一下WorkerState类，代码如下：
// 一个枚举类，表示Worker的状态
enum class WorkerState {
// 拥有了CPU令牌，可以执行纯CPU任务，也可以执行非纯CPU任务
CPU_ACQUIRED,
// 可以执行非纯CPU任务
BLOCKING,
// 当前已经暂停，一段时间后将终止，也有可能被再次使用
PARKING,
// 休眠状态，用于初始状态，只能执行自己本地任务
DORMANT,
// 终止状态，将不再被使用
TERMINATED
}2.Worker类的继承关系与全局变量
接下来对Worker类的继承关系以及其中重要的全局变量进行分析，代码如下：
// 继承自Thread类
// 私有化无参的构造方法
internal inner class Worker private constructor() : Thread() {
init {
// 标记为守护线程
isDaemon = true
}
// 当前线程在存储线程池线程的数组workers中的索引位置
@Volatile
var indexInArray = 0
set(index) {
// 设置线程名
name = "$schedulerName-worker-${if (index == 0) "TERMINATED" else index.toString()}"
field = index
}
// 构造方法
constructor(index: Int) : this() {
indexInArray = index
}
// 获取当前线程的调度器
inline val scheduler get() = this@CoroutineScheduler
// 线程存储任务的本地队列
@JvmField
val localQueue: WorkQueue = WorkQueue()
// 线程的状态 （内部转换）
@JvmField
var state = WorkerState.DORMANT
// 线程的控制状态（外部赋予）
val workerCtl = atomic(CLAIMED)
// 终止截止时间，表示处于PARKING状态的线程，在terminationDeadline毫秒后终止
private var terminationDeadline = 0L
// 表示当线程处于PARKING状态，进入parkedWorkersStack后，
// 下一个处于PARKING状态并进入parkedWorkersStack的线程的引用
@Volatile
var nextParkedWorker: Any? = NOT_IN_STACK
// 偷取其他线程的本地队列的任务的冷却时间，后面会解释
private var minDelayUntilStealableTaskNs = 0L
// 生成随机数，配合算法，用于任务寻找
private var rngState = Random.nextInt()
...
// 表示当前线程的本地队列是否有任务
@JvmField
var mayHaveLocalTasks = false
...
}3.Worker类的run方法
接下来分析Worker类的核心方法——run方法的实现，代码入下：
override fun run() = runWorker()
private fun runWorker() {
// 用于配合minDelayUntilStealableTaskNs自旋
var rescanned = false
// 线程池未关闭，线程没有终止，则循环
while (!isTerminated && state != WorkerState.TERMINATED) {
// 寻找并获取任务
val task = findTask(mayHaveLocalTasks)
// 如果找到了任务
if (task != null) {
// 重制两个变量
rescanned = false
minDelayUntilStealableTaskNs = 0L
// 执行任务
executeTask(task)
// 继续循环
continue
} else { // 如果没有找到任务，说明本地队列肯定没有任务，因为本地队列优先查找
// 设置标志位
mayHaveLocalTasks = false
}
// 走到这里，说明没有找到任务
// 如果偷取任务的冷却时间不为0，说明之前偷到过任务
if (minDelayUntilStealableTaskNs != 0L) {
// 这里通过rescanned，首次minDelayUntilStealableTaskNs不为0，
// 不会立刻进入PARKING状态，而是再次去寻找任务
// 因为当过多的线程进入PARKING状态，再次唤起大量的线程很难控制
if (!rescanned) {
rescanned = true
} else {// 再次扫描，仍然没有找到任务
// 置位
rescanned = false
// 尝试释放CPU令牌，并进入WorkerState.PARKING状态
tryReleaseCpu(WorkerState.PARKING)
// 清除中断标志位
interrupted()
// 阻塞minDelayUntilStealableTaskNs毫秒
LockSupport.parkNanos(minDelayUntilStealableTaskNs)
// 清零
minDelayUntilStealableTaskNs = 0L
}
// 阻塞完成后继续执行
continue
}
// 走到这里，说明线程可能很长时间都没有执行任务了，则对其进行暂停处理
// tryPark比tryReleaseCpu要严格的多，会被线程会被计入到parkedWorkersStack，
// 同时会修改workerCtl状态
tryPark()
}
// 退出循环
// 尝试释放CPU令牌，并进入终止状态
tryReleaseCpu(WorkerState.TERMINATED)
}4.Worker类的任务寻找机制
接下来分析Worker线程如何寻找任务，代码如下：
// 寻找任务
fun findTask(scanLocalQueue: Boolean): Task? {
// 尝试获取CPU令牌，如果获取到了，则调用findAnyTask方法，寻找任务
if (tryAcquireCpuPermit()) return findAnyTask(scanLocalQueue)
// 如果没有获取到CPU令牌，只能去找非纯CPU任务了
// 如果允许扫描本地的任务队列，则优先在本地队列中寻找，
// 找不到则在全局队列中寻找，从队首中获取
val task = if (scanLocalQueue) {
localQueue.poll() ?: globalBlockingQueue.removeFirstOrNull()
} else {
globalBlockingQueue.removeFirstOrNull()
}
// 如果在本地队列和全局队列中都找不到，则尝试去其他线程的队列里偷一个任务
return task ?: trySteal(blockingOnly = true)
}
// 寻找CPU任务
private fun findAnyTask(scanLocalQueue: Boolean): Task? {
// 如果允许扫描本地的任务队列，则在本地队列和全局队列中随机二选一，
// 找不到则在全局队列中寻找，从队首中获取
if (scanLocalQueue) {
// 随机确定本地队列和全局队列的优先顺序
val globalFirst = nextInt(2 * corePoolSize) == 0
// 获取任务
if (globalFirst) pollGlobalQueues()?.let { return it }
localQueue.poll()?.let { return it }
if (!globalFirst) pollGlobalQueues()?.let { return it }
} else {
// 只能从全局获取
pollGlobalQueues()?.let { return it }
}
// 走到这里，说明本地队列和全局队列中都找不到
// 那么就尝试去其他线程的队列里偷一个任务
return trySteal(blockingOnly = false)
}
// 从全局队列获取任务
private fun pollGlobalQueues(): Task? {
// 随机获取CPU任务或者非CPU任务
if (nextInt(2) == 0) {
// 优先获取CPU任务
globalCpuQueue.removeFirstOrNull()?.let { return it }
return globalBlockingQueue.removeFirstOrNull()
} else {
// 优先获取非CPU任务
globalBlockingQueue.removeFirstOrNull()?.let { return it }
return globalCpuQueue.removeFirstOrNull()
}
}
// 偷取其他线程的本地队列的任务
// blockingOnly表示是否只偷取阻塞任务
private fun trySteal(blockingOnly: Boolean): Task? {
// 只有当前线程的本地队列为空的时候，才能偷其他线程的本地队列
assert { localQueue.size == 0 }
// 获取已经存在的线程的数量
val created = createdWorkers
// 如果线程总数为0或1，则不偷取，直接返回
// 0:需要等待初始化
// 1:避免在单线程机器上过度偷取
if (created < 2) {
return null
}
// 随机生成一个存在的线程索引
var currentIndex = nextInt(created)
// 默认的偷取冷却时间
var minDelay = Long.MAX_VALUE
// 循环遍历
repeat(created) {
// 每次循环索引自增，带着下一行代码表示，从位置currentIndex开始偷
++currentIndex
// 如果超出了，则从头继续
if (currentIndex > created) currentIndex = 1
// 从数组中获取线程
val worker = workers[currentIndex]
// 如果线程不为空，并且不是自己
if (worker !== null && worker !== this) {
assert { localQueue.size == 0 }
// 根据偷取的类型进行偷取
val stealResult = if (blockingOnly) {
// 偷取非CPU任务到本地队列中
localQueue.tryStealBlockingFrom(victim = worker.localQueue)
} else {
// 偷取任务到本地队列中
localQueue.tryStealFrom(victim = worker.localQueue)
}
// 如果返回值为TASK_STOLEN，说明偷到了
// 如果返回值为NOTHING_TO_STEAL，说明要偷的线程的本地队列是空的
if (stealResult == TASK_STOLEN) {
// 从队列的队首拿出来返回
return localQueue.poll()
// 如果返回值大于零，表示偷取的冷却时间，说明没有偷到
} else if (stealResult > 0) { // 说明至少还要等待stealResult时间才能偷取这个任务
// 计算偷取冷却时间
minDelay = min(minDelay, stealResult)
}
}
}
// 设置偷取等待时间
minDelayUntilStealableTaskNs = if (minDelay != Long.MAX_VALUE) minDelay else 0
// 返回空
return null
}
// 基于Marsaglia xorshift RNG算法
// 用于在2^32-1范围内计算偷取目标
internal fun nextInt(upperBound: Int): Int {
var r = rngState
r = r xor (r shl 13)
r = r xor (r shr 17)
r = r xor (r shl 5)
rngState = r
val mask = upperBound - 1
// Fast path for power of two bound
if (mask and upperBound == 0) {
return r and mask
}
return (r and Int.MAX_VALUE) % upperBound
}通过对这部分代码的分析，可以知道线程在寻找任务时，首先会尝试获取CPU令牌，成为核心线程。如果线程成为了核心线程，则随机从本地或全局的两个队列中获取一个任务，获取不到则去随机偷取一个任务。如果没有获取到CPU令牌，则优先在本地获取任务，获取不到则在全局非CPU任务队列中获取任务，获取不到则去偷取一个非CPU任务。
如果偷取的任务没有达到最小的可偷取时间，则返回需要等待的时间。如果偷取任务成功，则直接加入到本地队列中。偷取的核心过程，会在后面进行分析。
5.Worker类的任务执行机制
接下来分析任务被获取到后如何被执行，代码如下：
// 执行任务
private fun executeTask(task: Task) {
// 获取任务类型，类型为纯CPU或可能阻塞
val taskMode = task.mode
// 重置线程闲置状态
idleReset(taskMode)
// 任务执行前
beforeTask(taskMode)
// 执行任务
runSafely(task)
// 任务执行后
afterTask(taskMode)
}
// 重置线程闲置状态
private fun idleReset(mode: Int) {
// 重置从PARKING状态到TERMINATED状态的时间
terminationDeadline = 0L
// 如果当前状态为PARKING，说明寻找任务时没有获取到CPU令牌
if (state == WorkerState.PARKING) {
assert { mode == TASK_PROBABLY_BLOCKING }
// 设置状态为BLOCKING
state = WorkerState.BLOCKING
}
}
// 任务执行前
private fun beforeTask(taskMode: Int) {
// 如果执行的任务为纯CPU任务，说明当前线程获取到了CPU令牌，是核心线程，直接返回
if (taskMode == TASK_NON_BLOCKING) return
// 走到这里，说明线程执行的是非纯CPU任务，
// 没有CPU令牌也可以执行，因此尝试释放CPU令牌，进入WorkerState.BLOCKING
if (tryReleaseCpu(WorkerState.BLOCKING)) {
// 如果释放CPU令牌成功，则唤起一个线程去申请CPU令牌
signalCpuWork()
}
}
// 执行任务
fun runSafely(task: Task) {
try {
task.run()
} catch (e: Throwable) {
// 异常发生时，通知当前线程的异常处理Handler
val thread = Thread.currentThread()
thread.uncaughtExceptionHandler.uncaughtException(thread, e)
} finally {
unTrackTask()
}
}
// 任务执行后
private fun afterTask(taskMode: Int) {
// 如果执行的任务为纯CPU任务，说明当前线程获取到了CPU令牌，是核心线程，直接返回
if (taskMode == TASK_NON_BLOCKING) return
// 如果执行的是非CPU任务
// 当前执行的非CPU任务数量减一
decrementBlockingTasks()
// 获取当前线程状态
val currentState = state
// 如果线程当前不是终止状态
if (currentState !== WorkerState.TERMINATED) {
assert { currentState == WorkerState.BLOCKING }
// 设置为休眠状态
state = WorkerState.DORMANT
}
}四.CoroutineScheduler类的dispatch方法
了解Worker类的工作机制后，接下来分析CoroutineScheduler类的dispatch方法，代码如下：
// block表示要执行的任务
// taskContext表示任务执行的上下文，里面包含任务的类型，和执行完成后的回调
// tailDispatch表示当前任务是否进行队列尾部调度，
// 当tailDispatch为true时，当前block会在当前线程的本地队列里的任务全部执行完后再执行
fun dispatch(block: Runnable, taskContext: TaskContext = NonBlockingContext, tailDispatch: Boolean = false) {
// 上报时间，TimeSource相关，无需关注
trackTask()
// 创建任务
val task = createTask(block, taskContext)
// 获取当前的Worker，可能获取不到
val currentWorker = currentWorker()
// 将当前的任务添加到当前线程的本地队列中
val notAdded = currentWorker.submitToLocalQueue(task, tailDispatch)
// 不为空，说明没有添加进去，说明当前的线程不是Worker
if (notAdded != null) {
// 将任务添加到全局队列中，如果添加失败了
if (!addToGlobalQueue(notAdded)) {
// 说明线程池正在关闭，抛出异常
throw RejectedExecutionException("$schedulerName was terminated")
}
}
// skipUnpark表示是否跳过唤起状态，取决于这下面两个参数
val skipUnpark = tailDispatch && currentWorker != null
// 如果当前类型为纯CPU任务
if (task.mode == TASK_NON_BLOCKING) {
// 如果跳过唤醒，则直接返回
if (skipUnpark) return
// 唤醒一个执行纯CPU任务的线程
signalCpuWork()
} else {
// 唤醒一个执行非CPU任务的线程
signalBlockingWork(skipUnpark = skipUnpark)
}
}
// 创建任务
internal fun createTask(block: Runnable, taskContext: TaskContext): Task {
// 获取当前时间
val nanoTime = schedulerTimeSource.nanoTime()
// 如果当前的block是Task类型的
if (block is Task) {
// 重新设置提交时间和任务上下文
block.submissionTime = nanoTime
block.taskContext = taskContext
// 返回
return block
}
// 封装成TaskImpl，返回
return TaskImpl(block, nanoTime, taskContext)
}
// 任务模型
// block表示执行的任务
// submissionTime表示任务提交时间
// taskContext表示任务执行的上下文
internal class TaskImpl(
@JvmField val block: Runnable,
submissionTime: Long,
taskContext: TaskContext
) : Task(submissionTime, taskContext) {
override fun run() {
try {
block.run()
} finally {
// 任务执行完毕后，会在同一个Worker线程中回调afterTask方法
taskContext.afterTask()
}
}
override fun toString(): String =
"Task[${block.classSimpleName}@${block.hexAddress}, $submissionTime, $taskContext]"
}
// 将任务添加到本地队列
private fun Worker?.submitToLocalQueue(task: Task, tailDispatch: Boolean): Task? {
// 如果当前线程为空，则返回任务
if (this == null) return task
// 如果线程处于终止状态，则返回任务
if (state === WorkerState.TERMINATED) return task
// 如果任务为纯CPU任务，但是线程没有CPU令牌
if (task.mode == TASK_NON_BLOCKING && state === WorkerState.BLOCKING) {
// 则返回任务
return task
}
// 标记本地队列有任务
mayHaveLocalTasks = true
// 添加到队列
return localQueue.add(task, fair = tailDispatch)
}
// 添加到全局队列
private fun addToGlobalQueue(task: Task): Boolean {
// 根据任务的类型，添加到全局队列的队尾
return if (task.isBlocking) {
globalBlockingQueue.addLast(task)
} else {
globalCpuQueue.addLast(task)
}
}
// 对当前线程进行强制转换，如果调度器也是当前的调度器则返回Worker对象
private fun currentWorker(): Worker? = (Thread.currentThread() as? Worker)?.takeIf { it.scheduler == this }
// 唤起一个执行非纯CPU任务的线程
private fun signalBlockingWork(skipUnpark: Boolean) {
// 当前执行阻塞任务的线程数量加1，并获取当前的控制状态
val stateSnapshot = incrementBlockingTasks()
// 如果跳过唤起，则返回
if (skipUnpark) return
// 尝试唤起，唤起成功，则返回
if (tryUnpark()) return
// 唤起失败，则根据当前的控制状态，尝试创建新线程，成功则返回
if (tryCreateWorker(stateSnapshot)) return
// 再次尝试唤起，防止多线程竞争情况下，上面的tryUnpark方法正好卡在线程释放CPU令牌与进入PARKING状态之间
// 因为线程先释放CPU令牌，后进入PARKING状态
tryUnpark()
}
// 唤起一个执行纯CPU任务的线程
internal fun signalCpuWork() {
// 尝试唤起，唤起成功，则返回
if (tryUnpark()) return
// 唤起失败，则尝试创建新线程，成功则返回
if (tryCreateWorker()) return
// 再次尝试唤起，防止多线程竞争情况下，上面的tryUnpark方法正好卡在线程释放CPU令牌与进入PARKING状态之间
// 因为线程先释放CPU令牌，后进入PARKING状态
tryUnpark()
}通过对上面的代码进行分析，可以知道CoroutineScheduler类的dispatch方法，首先会对任务进行封装。正常情况下，任务都会根据类型添加到全局队列中，接着根据任务类型，随机唤起一个执行对应类型任务的线程去执行任务。
当任务执行完毕后，会回调任务中自带的afterTask方法。根据之前对LimitingDispatcher的分析，可以知道，此时tailDispatch参数为true，同时当前的线程也是Worker线程，因此会被直接添加到线程的本地队列中，由于任务有对应的线程执行，因此跳过了唤起其他线程执行任务的阶段。这里我们可以称这个机制为尾调机制。
为什么CoroutineScheduler类中要设计一个尾调机制呢？
在传统的线程池的线程充足情况下，一个任务到来时，会被分配一个线程。假设前后两个任务A与B有依赖关系，需要在执行A再执行B，这时如果两个任务同时到来，执行A任务的线程会直接执行，而执行B线程的任务可能需要被阻塞。而一旦线程阻塞会造成线程资源的浪费。而协程本质上就是多个小段程序的相互协作，因此这种场景会非常多，通过这种机制可以保证任务的执行顺序，同时减少资源浪费，而且可以最大限度的保证一个连续的任务执行在同一个线程中。
至此，Dispatchers.IO线程池的工作原理全部分析完毕。
五.浅谈WorkQueue类
1.add方法
接下来分析一些更加细节的过程。首先分析一下Worker线程本地队列调用的add方法是如何添加任务的，代码如下：
// 本地队列中存储最后一次尾调的任务
private val lastScheduledTask = atomic<Task?>(null)
// fair表示是否公平的执行任务，FIFO，默认为false
fun add(task: Task, fair: Boolean = false): Task? {
// fair为true，则添加到队尾
if (fair) return addLast(task)
// 如果fair为false，则从lastScheduledTask中取出上一个尾调的任务，
// 并把这次的新尾调任务保存到lastScheduledTask
val previous = lastScheduledTask.getAndSet(task) ?: return null
// 如果获取上一次的尾调任务不为空，则添加到队尾
return addLast(previous)
}2.任务偷取机制
根据之前对Worker类的分析，任务偷取的核心代码锁定在了WorkQueue类的两个方法上：一个是偷取非纯CPU任务的tryStealBlockingFrom方法，另一个可以偷所有类型任务的tryStealFrom方法，代码如下：
internal const val BUFFER_CAPACITY_BASE = 7
internal const val BUFFER_CAPACITY = 1 shl BUFFER_CAPACITY_BASE // 1000 0000
internal const val MASK = BUFFER_CAPACITY - 1 // 0111 1111
// 存储任务的数组，最多存储128
private val buffer: AtomicReferenceArray<Task?> = AtomicReferenceArray(BUFFER_CAPACITY)
// producerIndex表示上一次向任务数组中添加任务的索引
// consumerIndex表示上一次消费的任务索引
// producerIndex永远大于等于consumerIndex
// 二者差值就是当前任务数组中任务的数量
private val producerIndex = atomic(0)
private val consumerIndex = atomic(0)
// buffer中非纯CPU任务的数量（避免遍历扫描）
private val blockingTasksInBuffer = atomic(0)
// 偷所有类型任务
fun tryStealFrom(victim: WorkQueue): Long {
assert { bufferSize == 0 }
// 从要偷取线程的本地队列中轮训获取一个任务
val task = victim.pollBuffer()
// 如果获取到了任务
if (task != null) {
// 将它添加到自己的本地队列中
val notAdded = add(task)
assert { notAdded == null }
// 返回偷取成功的标识
return TASK_STOLEN
}
// 如果偷取失败，尝试偷取指定线程的尾调任务
return tryStealLastScheduled(victim, blockingOnly = false)
}
// 轮训获取任务
private fun pollBuffer(): Task? {
// 死循环
while (true) {
// 获取上一次消费的任务索引
val tailLocal = consumerIndex.value
// 如果当前任务数组中没有多处的任务，则返回空
if (tailLocal - producerIndex.value == 0) return null
// 计算偷取位置，防止数组过界
val index = tailLocal and MASK
// 通过CAS方式，将consumerIndex加一，表示下一次要从tailLocal + 1处开始偷取
if (consumerIndex.compareAndSet(tailLocal, tailLocal + 1)) {
// 从偷取位置初取出任务，如果偷取的任务为空，则继续循环
val value = buffer.getAndSet(index, null) ?: continue
// 偷取成功
// 若任务为阻塞任务，blockingTasksInBuffer的值减一
value.decrementIfBlocking()
// 返回任务
return value
}
}
}
// 偷取非纯CPU任务
fun tryStealBlockingFrom(victim: WorkQueue): Long {
assert { bufferSize == 0 }
// 从consumerIndex位置开始偷
var start = victim.consumerIndex.value
// 偷到producerIndex处截止
val end = victim.producerIndex.value
// 获取任务数组
val buffer = victim.buffer
// 循环偷取
while (start != end) {
// 计算偷取位置，防止数组过界
val index = start and MASK
// 如果非纯CPU任务数为0，则直接退出循环
if (victim.blockingTasksInBuffer.value == 0) break
// 获取index处的任务
val value = buffer[index]
// 如果任务存在，而且是非纯CPU任务，同时成功的通过CAS设置为空
if (value != null && value.isBlocking && buffer.compareAndSet(index, value, null)) {
// blockingTasksInBuffer的值减一
victim.blockingTasksInBuffer.decrementAndGet()
// 将偷取的任务添加到当前线程的本地队列中
add(value)
// 返回偷取成功标识
return TASK_STOLEN
} else {
// 如果偷取失败，自增再次循环，从下一个位置开始偷
++start
}
}
// 如果从任务数组中偷取失败，尝试偷取指定线程的尾调任务
return tryStealLastScheduled(victim, blockingOnly = true)
}
// 偷取指定线程的尾调任务
private fun tryStealLastScheduled(victim: WorkQueue, blockingOnly: Boolean): Long {
// 死循环
while (true) {
// 获取指定线程的尾调任务，如果任务不存在，则返回偷取失败标识符
val lastScheduled = victim.lastScheduledTask.value ?: return NOTHING_TO_STEAL
// 如果要偷取的是非纯CPU任务，但是任务类型为纯CPU任务，说明只有核心线程才能偷
// 返回偷取失败标识符
if (blockingOnly && !lastScheduled.isBlocking) return NOTHING_TO_STEAL
// 获取当前时间
val time = schedulerTimeSource.nanoTime()
//计算任务从添加开始到现在经过的时长
val staleness = time - lastScheduled.submissionTime
// 如果时长小于偷取冷却时间
if (staleness < WORK_STEALING_TIME_RESOLUTION_NS) {
// 返回当前线程需要等待的时间
return WORK_STEALING_TIME_RESOLUTION_NS - staleness
}
// 通过CAS，将lastScheduledTask设置为空，防止被其他线程执行
if (victim.lastScheduledTask.compareAndSet(lastScheduled, null)) {
// 偷取成功，加入到当前线程的队列中
add(lastScheduled)
// 返回偷取成功表示
return TASK_STOLEN
}
// 继续循环
continue
}
}
// 偷取冷却时间，尾调任务从添加开始，
// 最少经过WORK_STEALING_TIME_RESOLUTION_NS时间才可以被偷
@JvmField
internal val WORK_STEALING_TIME_RESOLUTION_NS = systemProp(
"kotlinx.coroutines.scheduler.resolution.ns", 100000L
)六.总结
1.两个线程池
CoroutineScheduler类是核心的线程池，用于任务的执行。LimitingDispatcher类对CoroutineScheduler类进行代理，是CoroutineScheduler类尾调机制的使用者，对任务进行初步排队。
2.四种队列
LimitingDispatcher类中的任务队列。CoroutineScheduler类中的两个全局队列。Worker类中的本地队列。
3.尾调机制
一个任务执行完，可以通过回调，在同一个Worker线程中再存储一个待执行任务，该任务将在Worker线程本地队列目前已存在的任务，执行完毕后再执行。
4.任务分类与权限控制
所有任务分成纯CPU任务和非纯CPU任务两种，对应着核心线程和非核心线程。
所有线程在执行前都先尝试成为核心线程，核心线程可以从两种任务中任意选择执行，非核心线程只能执行非纯CPU任务。核心线程如果选择执行非纯CPU任务会变成非核心线程
5.任务偷取机制
WorkQueue类根据随机算法提供任务偷取机制，一个Worker线程可以从其他Worker线程的本地队列中偷取任务。
6.执行梳理图
加载全部内容