Kotlin协程

1、协程是什么

协程是编译器的能力，因为协程并不需要操作系统和硬件的支持(线程需要)，是编译器为了让开发者写代码更简单方便, 提供了一些关键字, 并在内部自动生成了一些字节码

线程和协程的目的差异

• 线程的目的是提高CPU资源使用率， 使多个任务得以并行的运行，是为了服务于机器的.
• 协程的目的是为了让多个任务之间更好的协作，主要体现在代码逻辑上，是为了服务开发者 (能提升资源的利用率, 但并不是原始目的)

线程和协程的调度差异

• 线程的调度是系统完成的，一般是抢占式的，根据优先级来分配
• 协程的调度是开发者根据程序逻辑指定好的，在不同的时期把资源合理的分配给不同的任务.

协程与线程的关系
协程并不是取代线程，而且抽象于线程之上，线程是被分割的CPU资源，协程是组织好的代码流程，协程需要线程来承载运行，线程是协程的资源

2、基本使用

CoroutineScope.launch

• launch函数可以启动新协程而不将结果返回给调用方

代码实现

//获取一个协程作用域用于创建协程
private val mScope = MainScope()

mScope.launch(Dispatchers.IO) {
            //IO线程执行getStringInfo()方法，返回结果
            var res = getStringInfo()
            //获取结果后主线程提示更新
            withContext(Dispatchers.Main) {
                Alerter.create(this@LearnCoroutineActivity).setTitle("Result").setText(res).show()
            }
}

private suspend fun getStringInfo(): String {
        return withContext(Dispatchers.IO) {
            //在这1000毫秒内该协程所处的线程不会阻塞
            delay(1000)
            "Coroutine-launch"
        }
}
    
//在onDestroy生命周期方法之中要手动取消
override fun onDestroy() {
        super.onDestroy()
        mScope.cancel()
}
复制代码

步骤

1. 获取一个协程作用域用于创建协程
2. 通过协程作用域.launch方法启动新的协程任务
   1. 启动时可以指定执行线程
   2. 内部通过withContext()方法实现切换线程
3. 在onDestroy生命周期方法之中要手动取消

协程作用域

• MainScope是协程默认提供的作用域，但是还有其他作用域更为方便
• 可使用lifecycleScope或者viewModelScope，这两种作用域会自动取消
• 在UI组件中使用LifecycleOwner.lifecycleScope，在ViewModel中使用ViewModel.viewModelScope

CoroutineScope.async

• async函数实现返回值处理或者并发处理

返回值处理

private fun asyncReturn() {
        mScope.launch(Dispatchers.Main) {
            //新开一个协程去执行协程体，父协程的代码会接着往下走
            var deferred = async(Dispatchers.IO) {
                delay(1000)
                "Coroutine-Async"
            }
            //等待async执行完成获取返回值，并不会阻塞线程，而是挂起，将线程的执行权交出去
            //直到async的协程体执行完毕后，会恢复协程继续执行
            val data = deferred.await()
            Alerter.create(this@LearnCoroutineActivity).setTitle("Result").setText(data).show()
        }
    }
复制代码

并发处理

private fun asyncConcurrent() {
    	//coroutineContext的创建下文会有分析
        var coroutineContext = Job() +
                Dispatchers.Main +
                CoroutineExceptionHandler { coroutineContext, throwable ->
                    Log.e(
                        "CoroutineException",
                        "CoroutineExceptionHandler: $throwable"
                    )
                } +
                CoroutineName("asyncConcurrent")
        mScope.launch(coroutineContext) {
            val job1 = async(Dispatchers.IO) {
                delay(1000)
                "job1-finish"
            }
            val job2 = async(Dispatchers.IO) {
                delay(2000)
                "job2-finish"
            }
            val job3 = async(Dispatchers.IO) {
                delay(500)
                "job3-finish"
            }
            //等待各job执行完 将结果合并
            Alerter.create(this@LearnCoroutineActivity).setTitle("Result")
                .setText("job1:${job1.await()},job2:${job2.await()},job3:${job3.await()}").show()
        }
    }
复制代码

可以看到，这就是同步代码风格编写异步代码的强大之处，下面开始源码解析

3、coroutineContext解析

• CoroutineContext是一个特殊的集合，同时包含了Map和Set的特点

• 集合内部的元素Element是根据key去对应（Map特点），但是不允许重复（Set特点）

• Element之间可以通过+号进行组合

• Element有如下四类，共同组成了CoroutineContext

  • Job：协程的唯一标识，用来控制协程的生命周期(new、active、completing、completed、cancelling、cancelled)
  • CoroutineDispatcher：指定协程运行的线程(IO、Default、Main、Unconfined)
  • CoroutineName: 指定协程的名称，默认为coroutine
  • CoroutineExceptionHandler: 指定协程的异常处理器，用来处理未捕获的异常

3.1、Job Element

• 每一个所创建的协程 (通过 launch 或者 async)，会返回一个 Job实例，该实例是协程的唯一标识，并且负责管理协程的生命周期

Job状态

Job在执行的过程中，包含了一系列状态，虽然开发者没办法直接获取所有状态，但是Job之中有如下三个属性

• isActive（是否活动）
• isCompleted（是否已完成）
• isCancelled（是否已取消）

根据属性就可以推断出Job的所处状态，状态如下

• 新创建 (New)
  • 当一个协程创建后就处于新建(New)状态
• 活跃 (Active)
  • 当调用Job的start/join方法后协程就处于活跃(Active)状态
• 完成中 (Completing)
  • 当协程执行完成后或者调用CompletableJob(CompletableJob是Job的一个子接口)的complete方法都会让当前协程进入完成中(Completing)状态
• 已完成 (Completed)
  • 处于完成中状态的协程会等所有子协程都完成后才进入完成(Completed)状态
• 取消中 (Cancelling)
  • 当运行出错或者调用Job的cancel方法都会将当前协程置为取消中(Cancelling)状态
• 已取消 (Cancelled)
  • 处于取消中状态的协程会等所有子协程都完成后才进入取消 (Cancelled)状态

State	isActive	isCompleted	isCancelled
New (optional initial state)	false	false	false
Active (default initial state)	true	false	false
Completing (transient state)	true	false	false
Cancelling (transient state)	false	false	true
Cancelled (final state)	false	true	true
Completed (final state)	false	true	false

                                      wait children
+-----+ start  +--------+ complete   +-------------+  finish  +-----------+
| New | -----> | Active | ---------> | Completing  | -------> | Completed |
+-----+        +--------+            +-------------+          +-----------+
                 |  cancel / fail       |
                 |     +----------------+
                 |     |
                 V     V
             +------------+                           finish  +-----------+
             | Cancelling | --------------------------------> | Cancelled |
             +------------+                                   +-----------+

复制代码

Job方法

fun start(): Boolean

• 调用该函数来启动这个 Coroutine，如果当前 Coroutine 还没有执行调用该函数返回 true，如果当前 Coroutine 已经执行或者已经执行完毕，则调用该函数返回 false

fun cancel(cause: CancellationException? = null)

• 通过可选的取消原因取消Job

fun invokeOnCompletion(handler: CompletionHandler): DisposableHandle

• 通过这个函数可以给 Job 设置一个完成通知，当 Job 执行完成的时候会同步执行这个通知函数。 回调的通知对象类型为：typealias CompletionHandler = (cause: Throwable?) -> Unit.
• CompletionHandler 参数代表了 Job 是如何执行完成的。 cause 有下面三种情况：
  • 如果 Job 是正常执行完成的，则 cause 参数为 null
  • 如果 Job 是正常取消的，则 cause 参数为 CancellationException 对象。这种情况不应该当做错误处理，这是任务正常取消的情形。所以一般不需要在错误日志中记录这种情况。
  • 其他情况表示 Job 执行失败了。
• 这个函数的返回值为 DisposableHandle 对象，如果不再需要监控 Job 的完成情况了， 则可以调用 DisposableHandle.dispose 函数来取消监听。如果 Job 已经执行完了， 则无需调用 dispose 函数了，会自动取消监听。

suspend fun join()（suspend函数）

• 用来在另外一个 Coroutine 中等待 job 执行完成后继续执行。

Job异常传播

• 协程是有父子级的概念，如果子Job在运行过程之中发生异常，那么父Job就会感知到并抛出异常。如果要抑制这种行为就需要使用SupervisorJob

  除了CancellationException以外的异常

SupervisorJob

fun main(){
     val parentJob = GlobalScope.launch {
       //childJob是一个SupervisorJob
        val childJob = launch(SupervisorJob()){
            throw NullPointerException()
        }
        childJob.join()
        println("parent complete")
    }
    Thread.sleep(1000)
}
复制代码

此时childJob抛出异常并不会影响parentJob的运行，parentJob会继续运行并输出parent complete。

3.2、CoroutineDispatcher Element

• 用于指定协程的运行线程
• kotlin已经内置了CoroutineDispatcher的4个实现，可以通过Dispatchers的Default、IO、Main、Unconfined字段分别返回使用

public actual object Dispatchers {
    @JvmStatic
    public actual val Default: CoroutineDispatcher = createDefaultDispatcher()
    @JvmStatic
    public val IO: CoroutineDispatcher = DefaultScheduler.IO
    @JvmStatic
    public actual val Unconfined: CoroutineDispatcher = kotlinx.coroutines.Unconfined
    @JvmStatic
    public actual val Main: MainCoroutineDispatcher get() = MainDispatcherLoader.dispatcher
}
复制代码

Default，IO

Default，IO其实内部用的是一个线程池，下面逐个解析，看实现原理

default

• Default会根据useCoroutinesScheduler属性（默认为true）去获取对应的线程池
  • DefaultScheduler（useCoroutinesScheduler=ture）：kotlin自己实现的线程池逻辑
  • CommonPool（useCoroutinesScheduler=false）：java类库中的Executor实现线程池逻辑

internal actual fun createDefaultDispatcher(): CoroutineDispatcher =
    if (useCoroutinesScheduler) DefaultScheduler else CommonPool
internal object DefaultScheduler : ExperimentalCoroutineDispatcher() {
    .....
}
//委托类
public open class ExperimentalCoroutineDispatcher(
    private val corePoolSize: Int,
    private val maxPoolSize: Int,
    private val idleWorkerKeepAliveNs: Long,
    private val schedulerName: String = "CoroutineScheduler"
) : ExecutorCoroutineDispatcher() {
}
//java类库中的Executor实现线程池逻辑
internal object CommonPool : ExecutorCoroutineDispatcher() {}
//共同父类，定义行为
public abstract class ExecutorCoroutineDispatcher: CoroutineDispatcher(), Closeable {}
复制代码

ExperimentalCoroutineDispatcher

• DefaultScheduler的主要实现都在它的父类ExperimentalCoroutineDispatcher中

public open class ExperimentalCoroutineDispatcher(
    private val corePoolSize: Int,
    private val maxPoolSize: Int,
    private val idleWorkerKeepAliveNs: Long,
    private val schedulerName: String = "CoroutineScheduler"
) : ExecutorCoroutineDispatcher() {
    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 fun createScheduler() = CoroutineScheduler(corePoolSize, maxPoolSize, idleWorkerKeepAliveNs, schedulerName)
    
    override val executor: Executorget() = coroutineScheduler

    override fun dispatch(context: CoroutineContext, block: Runnable): Unit =
        try {
            //dispatch方法委托到CoroutineScheduler的dispatch方法
            coroutineScheduler.dispatch(block)
        } catch (e: RejectedExecutionException) {
            ....
        }

    override fun dispatchYield(context: CoroutineContext, block: Runnable): Unit =
        try {
            //dispatchYield方法委托到CoroutineScheduler的dispatchYield方法
            coroutineScheduler.dispatch(block, tailDispatch = true)
        } catch (e: RejectedExecutionException) {
            ...
        }
    
	internal fun dispatchWithContext(block: Runnable, context: TaskContext, tailDispatch: Boolean) {
        try {
            //dispatchWithContext方法委托到CoroutineScheduler的dispatchWithContext方法
            coroutineScheduler.dispatch(block, context, tailDispatch)
        } catch (e: RejectedExecutionException) {
            ....
        }
    }
    override fun close(): Unit = coroutineScheduler.close()
    //实现请求阻塞
    public fun blocking(parallelism: Int = BLOCKING_DEFAULT_PARALLELISM): CoroutineDispatcher {
        require(parallelism > 0) { "Expected positive parallelism level, but have $parallelism" }
        return LimitingDispatcher(this, parallelism, null, TASK_PROBABLY_BLOCKING)
    }
	//实现请求数量限制
    public fun limited(parallelism: Int): CoroutineDispatcher {
        require(parallelism > 0) { "Expected positive parallelism level, but have $parallelism" }
        require(parallelism <= corePoolSize) { "Expected parallelism level lesser than core pool size ($corePoolSize), but have $parallelism" }
        return LimitingDispatcher(this, parallelism, null, TASK_NON_BLOCKING)
    }
    
    ....//省略一些供测试的方法，更好的跟踪同步状态
}
复制代码

IO

• IO的实现其实是LimitingDispatcher

val IO: CoroutineDispatcher = LimitingDispatcher(
    this,
    systemProp(IO_PARALLELISM_PROPERTY_NAME, 64.coerceAtLeast(AVAILABLE_PROCESSORS)),
    "Dispatchers.IO",
    TASK_PROBABLY_BLOCKING
)
复制代码

LimitingDispatcher

• IO的实现类会有一些最大请求限制，以及队列处理

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>()
    //cas计数
    private val inFlightTasks = atomic(0)
    
    override fun dispatch(context: CoroutineContext, block: Runnable) = dispatch(block, false)

    private fun dispatch(block: Runnable, tailDispatch: Boolean) {
        var taskToSchedule = block
        while (true) {

            if (inFlight <= parallelism) {
                //LimitingDispatcher的dispatch方法委托给了DefaultScheduler的dispatchWithContext方法
                dispatcher.dispatchWithContext(taskToSchedule, this, tailDispatch)
                return
            }
            ..//省略了一些队列处理逻辑
        }
    }
}
复制代码

CoroutineScheduler

• Default、IO其实都是共享CoroutineScheduler线程池，Kotlin实现了一套线程池两种调度策略
• 通过内部的mode区分

fun dispatch(block: Runnable, taskContext: TaskContext = NonBlockingContext, tailDispatch: Boolean = false) {
......
    if (task.mode == TASK_NON_BLOCKING) {
        if (skipUnpark) return
        signalCpuWork()
    } else {
        signalBlockingWork(skipUnpark = skipUnpark)
    }
}
复制代码

Mode

Type	Mode
Default	TASK_NON_BLOCKING
IO	TASK_PROBABLY_BLOCKING

处理策略

Type	Mode
Default	CoroutineScheduler最多有corePoolSize个线程被创建，corePoolSize它的取值为max(2, CPU核心数)， 即它会尽量的等于CPU核心数
IO	创建比corePoolSize更多的线程来运行IO型任务，但不能大于maxPoolSize 1.公式：max(corePoolSize, min(CPU核心数 * 128, 2^21 – 2))，即大于corePoolSize，小于2^21 – 2 2.2^21 – 2是一个很大的数约为2M，但是CoroutineScheduler是不可能创建这么多线程的，所以就需要外部限制提交的任务数 3.Dispatchers.IO构造时就通过LimitingDispatcher默认限制了最大线程并发数parallelism为max(64, CPU核心数)，即最多只能提交parallelism个任务到CoroutineScheduler中执行，剩余的任务被放进队列中等待。

适合场景

Type	Mode
Default	1.CPU密集型任务的特点是执行任务时CPU会处于忙碌状态，任务会消耗大量的CPU资源 2.复杂计算、视频解码等，如果此时线程数太多，超过了CPU核心数，那么这些超出来的线程是得不到CPU的执行的，只会浪费内存资源 3.因为线程本身也有栈等空间，同时线程过多，频繁的线程切换带来的消耗也会影响线程池的性能 4.对于CPU密集型任务，线程池并发线程数等于CPU核心数才能让CPU的执行效率最大化
IO	1.IO密集型任务的特点是执行任务时CPU会处于闲置状态，任务不会消耗大量的CPU资源 2.网络请求、IO操作等，线程执行IO密集型任务时大多数处于阻塞状态，处于阻塞状态的线程是不占用CPU的执行时间 3.此时CPU就处于闲置状态，为了让CPU忙起来，执行IO密集型任务时理应让线程的创建数量更多一点，理想情况下线程数应该等于提交的任务数，对于这些多创建出来的线程，当它们闲置时，线程池一般会有一个超时回收策略，所以大部分情况下并不会占用大量的内存资源 4.但也会有极端情况，所以对于IO密集型任务，线程池并发线程数应尽可能地多才能提高CPU的吞吐量，这个尽可能地多的程度并不是无限大，而是根据业务情况设定，但肯定要大于CPU核心数。

Unconfined

• 任务执行在默认的启动线程。之后由调用resume的线程决定恢复协程的线程。

internal object Unconfined : CoroutineDispatcher() {
    //为false为不需要dispatch
    override fun isDispatchNeeded(context: CoroutineContext): Boolean = false

    override fun dispatch(context: CoroutineContext, block: Runnable) {
        // 只有当调用yield方法时，Unconfined的dispatch方法才会被调用
        // yield() 表示当前协程让出自己所在的线程给其他协程运行
        val yieldContext = context[YieldContext]
        if (yieldContext != null) {
            yieldContext.dispatcherWasUnconfined = true
            return
        }
        throw UnsupportedOperationException("Dispatchers.Unconfined.dispatch function can only be used by the yield function. " +
            "If you wrap Unconfined dispatcher in your code, make sure you properly delegate " +
            "isDispatchNeeded and dispatch calls.")
    }
}
复制代码

• 每一个协程都有对应的Continuation实例，其中的resumeWith用于协程的恢复，存在于DispatchedContinuation

DispatchedContinuation

• 我们重点看resumeWith的实现以及类委托

internal class DispatchedContinuation<in T>(
    @JvmField val dispatcher: CoroutineDispatcher,
    @JvmField val continuation: Continuation<T>
) : DispatchedTask<T>(MODE_UNINITIALIZED), CoroutineStackFrame, Continuation<T> by continuation {
    .....
    override fun resumeWith(result: Result<T>) {
        val context = continuation.context
        val state = result.toState()
        if (dispatcher.isDispatchNeeded(context)) {
            _state = state
            resumeMode = MODE_ATOMIC
            dispatcher.dispatch(context, this)
        } else {
            executeUnconfined(state, MODE_ATOMIC) {
                withCoroutineContext(this.context, countOrElement) {
                    continuation.resumeWith(result)
                }
            }
        }
    }
    ....
}
复制代码

解析如下：

1. DispatchedContinuation通过类委托实现了在resumeWith()方法之前的代码逻辑添加

2. 通过isDispatchNeeded（是否需要dispatch，Unconfined=false，default，IO=true）判断做不同处理

   1. true：调用协程的CoroutineDispatcher的dispatch方法（上文有解析）
   2. false：调用executeUnconfined方法

   private inline fun DispatchedContinuation<*>.executeUnconfined(
       contState: Any?, mode: Int, doYield: Boolean = false,
       block: () -> Unit
   ): Boolean {
       assert { mode != MODE_UNINITIALIZED }
       val eventLoop = ThreadLocalEventLoop.eventLoop
       if (doYield && eventLoop.isUnconfinedQueueEmpty) return false
       return if (eventLoop.isUnconfinedLoopActive) {
           _state = contState
           resumeMode = mode
           eventLoop.dispatchUnconfined(this)
           true
       } else {
           runUnconfinedEventLoop(eventLoop, block = block)
           false
       }
   }
   复制代码

   1. 从threadlocal中取出eventLoop（eventLoop和当前线程相关的），判断是否在执行Unconfined任务
      1. 如果在执行则调用EventLoop的dispatchUnconfined方法把Unconfined任务放进EventLoop中
      2. 如果没有在执行则直接执行

   internal inline fun DispatchedTask<*>.runUnconfinedEventLoop(
       eventLoop: EventLoop,
       block: () -> Unit
   ) {
       eventLoop.incrementUseCount(unconfined = true)
       try {
           block()
           while (true) {
               if (!eventLoop.processUnconfinedEvent()) break
           }
       } catch (e: Throwable) {
           handleFatalException(e, null)
       } finally {
           eventLoop.decrementUseCount(unconfined = true)
       }
   }
   复制代码

   1. 执行block()代码块，即上文提到的resumeWith()
   2. 调用processUnconfinedEvent()方法实现执行剩余的Unconfined任务，知道全部执行完毕跳出循环

EventLoop

• EventLoop是存放与threadlocal，所以是跟当前线程相关联的，而EventLoop也是CoroutineDispatcher的一个子类

internal abstract class EventLoop : CoroutineDispatcher() {
  	.....
    //双端队列实现存放Unconfined任务
    private var unconfinedQueue: ArrayQueue<DispatchedTask<*>>? = null
    //从队列的头部移出Unconfined任务执行
    public fun processUnconfinedEvent(): Boolean {
        val queue = unconfinedQueue ?: return false
        val task = queue.removeFirstOrNull() ?: return false
        task.run()
        return true
    }
    //把Unconfined任务放进队列的尾部
    public fun dispatchUnconfined(task: DispatchedTask<*>) {
        val queue = unconfinedQueue ?:
            ArrayQueue<DispatchedTask<*>>().also { unconfinedQueue = it }
        queue.addLast(task)
    }
    .....
}
复制代码

解析如下：

1. 内部通过双端队列实现存放Unconfined任务
   1. EventLoop的dispatchUnconfined方法用于把Unconfined任务放进队列的尾部
   2. rocessUnconfinedEvent方法用于从队列的头部移出Unconfined任务执行

Main

• 是把协程运行在平台相关的只能操作UI对象的Main线程，但是根据不同平台有不同的实现

平台	实现
kotlin/js	kotlin对JavaScript的支持，提供了转换kotlin代码，kotlin标准库的能力，npm包管理能力 在kotlin/js上Dispatchers.Main等效于Dispatchers.Default
kotlin/native	将kotlin代码编译为无需虚拟机就可运行的原生二进制文件的技术, 它的主要目的是允许对不需要或不可能使用虚拟机的平台进行编译，例如嵌入式设备或iOS 在kotlin/native上Dispatchers.Main等效于Dispatchers.Default
kotlin/JVM	需要虚拟机才能编译的平台，例如Android就是属于kotlin/JVM，对于kotlin/JVM我们需要引入对应的dispatcher，例如Android就需要引入kotlinx-coroutines-android库，它里面有Android对应的Dispatchers.Main实现，其实就是把任务通过Handler运行在Android的主线程

3.3、CoroutineName Element

• 协程名称，可以自定义，方便调试分析

public data class CoroutineName(
    val name: String
) : AbstractCoroutineContextElement(CoroutineName) {
 
    public companion object Key : CoroutineContext.Key<CoroutineName>
    
    override fun toString(): String = "CoroutineName($name)"
}
复制代码

3.4、CoroutineExceptionHandler Element

• 协程异常处理器，默认创建的协程都会有一个异常处理器，也可以手动指定。

  var coroutineContext = Job() +
          Dispatchers.Main +
  	   //手动添加指定异常处理器
          CoroutineExceptionHandler { coroutineContext, throwable ->
              Log.e(
                  "CoroutineException",
                  "CoroutineExceptionHandler: $throwable"
              )
          } +
          CoroutineName("asyncConcurrent")
  复制代码

• 但是只对launch方法启动的根协程有效，而对async启动的根协程无效

  async启动的根协程默认会捕获所有未捕获异常并把它放在Deferred中，等到用户调用Deferred的await方法才抛出，也就是需要手动加try-catch

CASE

协程的使用场景变化自如，异常处理的情况也就比较多

1. 非SupervisorJob情况下，字协程抛出的异常会委托给父协程的CoroutineExceptionHandler处理，子协程的CoroutineExceptionHandler并不会执行
2. SupervisorJob情况下，不会产生异常传播，即自己的CoroutineExceptionHandler可以接收到异常
3. 子协程同时抛出多个异常时，CoroutineExceptionHandler只会捕捉第一个异常，后续的异常存于第一个异常的suppressed数组之中
4. 取消协程时会抛出CancellationException，但是所有的CoroutineExceptionHandler不会接收到，只能通过try-catch实现捕获

3.5、CoroutineContext结构

• CoroutineContext是一个特殊的集合，同时包含了Map和Set的特点
• 集合内部的元素Element是根据key去对应（Map特点），但是不允许重复（Set特点）
• Element之间可以通过+号进行组合
• 每一个Element都继承与CoroutineContext

public interface CoroutineContext {
    //操作符[]重载，可以通过CoroutineContext[Key]来获取与Key关联的Element
    public operator fun <E : Element> get(key: Key<E>): E?

    //它是一个聚集函数，提供了从left到right遍历CoroutineContext中每一个Element的能力，并对每一个Element做operation操作
    public fun <R> fold(initial: R, operation: (R, Element) -> R): R
    
 	//操作符+重载，可以CoroutineContext + CoroutineContext这种形式把两个CoroutineContext合并成一个
    public operator fun plus(context: CoroutineContext): CoroutineContext =
        if (context === EmptyCoroutineContext) this else 
            context.fold(this) { acc, element ->
                val removed = acc.minusKey(element.key)
                if (removed === EmptyCoroutineContext) element else {
                    val interceptor = removed[ContinuationInterceptor]
                    if (interceptor == null) CombinedContext(removed, element) else {
                        val left = removed.minusKey(ContinuationInterceptor)
                        if (left === EmptyCoroutineContext) CombinedContext(element, interceptor) else
                            CombinedContext(CombinedContext(left, element), interceptor)
                    }
                }
            }
    
    //返回一个新的CoroutineContext，这个CoroutineContext删除了Key对应的Element
    public fun minusKey(key: Key<*>): CoroutineContext

    //Key定义，空实现
    public interface Key<E : Element>
    
    //Element定义，每个Element都是一个CoroutineContext
    public interface Element : CoroutineContext {
        .....
    }
}
复制代码

这里使用了Kotlin的operator操作符重载实现了各种策略

• 除了plus方法，CoroutineContext中的其他三个方法都被CombinedContext、Element、EmptyCoroutineContext重写
  • CombinedContext：CoroutineContext集合结构的实现，它里面是一个递归定义
  • Element：CombinedContext中的元素
  • EmptyCoroutineContext：一个空的CoroutineContext，它里面是空实现

CombinedContext

internal class CombinedContext(
    private val left: CoroutineContext,
    private val element: Element
) : CoroutineContext, Serializable {
    ....
}
复制代码

• CombinedContext包含left和element元素
  • left可能是element或者是CombinedContext
  • element就是element

get方法

• 可以通过CoroutineContext[Key]来获取与Key关联的Element

override fun <E : Element> get(key: Key<E>): E? {
    var cur = this
    while (true) {
        //element是否匹配，如果是则直接返回，即匹配成功
        cur.element[key]?.let { return it }
        //没有匹配成功则从left开始寻找
        val next = cur.left
        //如果left是CombinedContext，则改变next，重复上述步骤
        if (next is CombinedContext) {
            cur = next
        } else {
            //匹配成功则返回
            return next[key]
        }
    }
}
复制代码

fold方法

• 提供了从left到right遍历CoroutineContext中每一个Element的能力，并对每一个Element做operation操作

//operation是一个函数指针，可以执行函数引用
public override fun <R> fold(initial: R, operation: (R, Element) -> R): R =
    //对left做fold操作，把left做完fold操作的的返回结果和element做operation操作
    operation(left.fold(initial, operation), element)
复制代码

minusKey方法

• 返回一个新的CoroutineContext，这个CoroutineContext删除了Key对应的Element

public override fun minusKey(key: Key<*>): CoroutineContext {
        //element是否匹配，如果是则直接返回供删除，即匹配成功
        element[key]?.let { return left }
        //没有匹配成功则从left开始寻找
        val newLeft = left.minusKey(key)
        return when {
            //如果left中不存在目标element，则当前CombinedContext肯定不包含目标元素，直接返回当前
            newLeft === left -> this
            //如果left之中存在目标element，删除目标element后，left等于空，返回当前CombinedContext的element
            newLeft === EmptyCoroutineContext -> element
            //如果left之中存在目标element，删除目标element后，left不等于空，创建新的CombinedContext并返回
            else -> CombinedContext(newLeft, element)
        }
}
复制代码

结构图

• 整体像链表，left就是指向下一个结点的指针，
• get、minusKey操作逻辑流程都是先访问当前element，不满足，再访问left的element，顺序都是从right到left
• fold的操作逻辑流程是先访问left，直到递归到最后的element，然后再从left到right的返回，从而访问了所有的element。

plus方法

此方法是CoroutineContext的实现，内部分为元素合并和拦截器处理

public operator fun plus(context: CoroutineContext): CoroutineContext =
	    //如果合并的元素为空，直接返回
        if (context === EmptyCoroutineContext) this else 
		   //对要合并的元素座flod处理
            context.fold(this) { acc, element -> //acc==原有元素，element==合并元素
                //从原有元素之中尝试删除合并元素
                val removed = acc.minusKey(element.key)
                //如果删除元素为空则说明原有元素删除了和目标元素的相同元素后为空，则直接返回目标元素即可
                if (removed === EmptyCoroutineContext) element else {
                    //尝试获取对应的Interceptor
                    val interceptor = removed[ContinuationInterceptor]
                    //如果Interceptor为空则构建一个新的元素
                    if (interceptor == null) CombinedContext(removed, element) else {
                        //如果不为空则做删除操作
                        val left = removed.minusKey(ContinuationInterceptor)
                        if (left === EmptyCoroutineContext) CombinedContext(element, interceptor) else
                        	//如果删除元素为空则说明原有元素删除了和目标元素的相同元素后不为空，则构建一个新的元素并返回，并添加Interceptor至末尾
                            CombinedContext(CombinedContext(left, element), interceptor)
                    }
                }
            }
复制代码

• puls方法最终返回的CoroutineContext是不存在key相同的element的，新增元素会覆盖CoroutineContext中的含有相同key的元素，这像是Set的特性

• 拦截器的处理就是为了每次添加完成后保持ContinuationInterceptor为CoroutineContext中的最后一个元素，目的是在执行协程之前做前置操作

  CoroutineDispatcher就继承自ContinuationInterceptor

  • 通过把ContinuationInterceptor放在最后面，协程在查找上下文的element时，总能最快找到拦截器，避免了递归查找，从而让拦截行为前置执行