亲宝软件园·资讯

展开

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.执行梳理图

加载全部内容

相关教程
猜你喜欢
用户评论