协程及Kotlin协程

一、协程是什么？

协程是程序自己控制挂起和恢复的程序。

协程可以实现多任务协作执行。

二、协程作用？

• 协程可以让异步代码同步化。
• 协程可以降低异步程序的设计复杂度。

三、协程分类

• 按调用栈分类：

  • 有栈协程：每个协程都会分配一个单独调用栈，类似于线程的调用栈。
  • 无栈协程：协程不会分配一个单独调用栈，挂起点的状态通过闭包和对象保存。

• 按调用关系分类：

  • 对称协程：协程调用权可以转移给任意协程，协程之间是对等关系。比如：协程A调度协程B，协程B调度协程C，协程C调度协程A。
  • 非对称协程：调度权只能转移给调度自己的协程，协程存在父子关系。

四、Kotlin中协程

Kotlin实现的协程是无栈协程，挂起状态通过Continuation保存，CoroutineContext包含一个协程调度器。

1. 挂起函数及suspend修饰符

通过suspend修饰的函数就是挂起函数。

挂起函数只能通过其它挂起函数或者协程调用。

挂起函数调用时包含了协程的挂起语义。 

Continuation源码：

/**
 * Interface representing a continuation after a suspension point that returns a value of type `T`.
 */
@SinceKotlin("1.3")
public interface Continuation<in T> {
    /**
     * The context of the coroutine that corresponds to this continuation.
     */
    public val context: CoroutineContext

    /**
     * Resumes the execution of the corresponding coroutine passing a successful or failed [result] as the
     * return value of the last suspension point.
     */
    public fun resumeWith(result: Result<T>)
}

/**
 * Classes and interfaces marked with this annotation are restricted when used as receivers for extension
 * `suspend` functions. These `suspend` extensions can only invoke other member or extension `suspend` functions on this particular
 * receiver and are restricted from calling arbitrary suspension functions.
 */
@SinceKotlin("1.3")
@Target(AnnotationTarget.CLASS)
@Retention(AnnotationRetention.BINARY)
public annotation class RestrictsSuspension

/**
 * Resumes the execution of the corresponding coroutine passing [value] as the return value of the last suspension point.
 */
@SinceKotlin("1.3")
@InlineOnly
public inline fun <T> Continuation<T>.resume(value: T): Unit =
    resumeWith(Result.success(value))

/**
 * Resumes the execution of the corresponding coroutine so that the [exception] is re-thrown right after the
 * last suspension point.
 */
@SinceKotlin("1.3")
@InlineOnly
public inline fun <T> Continuation<T>.resumeWithException(exception: Throwable): Unit =
    resumeWith(Result.failure(exception))


/**
 * Creates a [Continuation] instance with the given [context] and implementation of [resumeWith] method.
 */
@SinceKotlin("1.3")
@InlineOnly
public inline fun <T> Continuation(
    context: CoroutineContext,
    crossinline resumeWith: (Result<T>) -> Unit
): Continuation<T> =
    object : Continuation<T> {
        override val context: CoroutineContext
            get() = context

        override fun resumeWith(result: Result<T>) =
            resumeWith(result)
    }

/**
 * Creates a coroutine without a receiver and with result type [T].
 * This function creates a new, fresh instance of suspendable computation every time it is invoked.
 *
 * To start executing the created coroutine, invoke `resume(Unit)` on the returned [Continuation] instance.
 * The [completion] continuation is invoked when the coroutine completes with a result or an exception.
 * Subsequent invocation of any resume function on the resulting continuation will produce an [IllegalStateException].
 */
@SinceKotlin("1.3")
@Suppress("UNCHECKED_CAST")
public fun <T> (suspend () -> T).createCoroutine(
    completion: Continuation<T>
): Continuation<Unit> =
    SafeContinuation(createCoroutineUnintercepted(completion).intercepted(), COROUTINE_SUSPENDED)

/**
 * Creates a coroutine with receiver type [R] and result type [T].
 * This function creates a new, fresh instance of suspendable computation every time it is invoked.
 *
 * To start executing the created coroutine, invoke `resume(Unit)` on the returned [Continuation] instance.
 * The [completion] continuation is invoked when the coroutine completes with a result or an exception.
 * Subsequent invocation of any resume function on the resulting continuation will produce an [IllegalStateException].
 */
@SinceKotlin("1.3")
@Suppress("UNCHECKED_CAST")
public fun <R, T> (suspend R.() -> T).createCoroutine(
    receiver: R,
    completion: Continuation<T>
): Continuation<Unit> =
    SafeContinuation(createCoroutineUnintercepted(receiver, completion).intercepted(), COROUTINE_SUSPENDED)

/**
 * Starts a coroutine without a receiver and with result type [T].
 * This function creates and starts a new, fresh instance of suspendable computation every time it is invoked.
 * The [completion] continuation is invoked when the coroutine completes with a result or an exception.
 */
@SinceKotlin("1.3")
@Suppress("UNCHECKED_CAST")
public fun <T> (suspend () -> T).startCoroutine(
    completion: Continuation<T>
) {
    createCoroutineUnintercepted(completion).intercepted().resume(Unit)
}

/**
 * Starts a coroutine with receiver type [R] and result type [T].
 * This function creates and starts a new, fresh instance of suspendable computation every time it is invoked.
 * The [completion] continuation is invoked when the coroutine completes with a result or an exception.
 */
@SinceKotlin("1.3")
@Suppress("UNCHECKED_CAST")
public fun <R, T> (suspend R.() -> T).startCoroutine(
    receiver: R,
    completion: Continuation<T>
) {
    createCoroutineUnintercepted(receiver, completion).intercepted().resume(Unit)
}

/**
 * Obtains the current continuation instance inside suspend functions and suspends
 * the currently running coroutine.
 *
 * In this function both [Continuation.resume] and [Continuation.resumeWithException] can be used either synchronously in
 * the same stack-frame where the suspension function is run or asynchronously later in the same thread or
 * from a different thread of execution. Subsequent invocation of any resume function will produce an [IllegalStateException].
 */
@SinceKotlin("1.3")
@InlineOnly
public suspend inline fun <T> suspendCoroutine(crossinline block: (Continuation<T>) -> Unit): T =
    suspendCoroutineUninterceptedOrReturn { c: Continuation<T> ->
        val safe = SafeContinuation(c.intercepted())
        block(safe)
        safe.getOrThrow()
    }

/**
 * Returns the context of the current coroutine.
 */
@SinceKotlin("1.3")
@Suppress("WRONG_MODIFIER_TARGET")
@InlineOnly
public suspend inline val coroutineContext: CoroutineContext
    get() {
        throw NotImplementedError("Implemented as intrinsic")
    }

 挂起函数格式：

suspend fun test(param: Int): String {
   return "test"
}

挂起函数类型：

suspend (param: Int) -> String

在编译后，解码挂起函数：

suspend fun test(param: Int, continuation: Continuation<String>): String {
    return "test"
}

在函数参数中多了一个Continuation参数，Continuation就是用于保存协程挂起点状态对象。这也是为什么挂起函数只能由其它挂起函数或者协程调用，为什么Kotlin编译器在编译时给挂起函数多加了一个参数。

2. 挂起与未挂起

• 真正的挂起：就是必须异步调用resume，包括：1. 切换到其它线程的resume；2. 单线程事件循环异步执行；
• 未挂起：就是未异步调用resume，直接将结果传递出去。

PS：在resume时返回值分两类，1. 函数真正的返回值，比如上面test函数，返回值类型为String类型；2. 返回挂起函数的挂起标识；只有返回挂起标识时才是真正的挂起。

3. 将回调转写成挂起函数：

• 使用suspendCoroutine获取挂起函数的Continuation；
• 回调成功分支使用Continuation.resume(value)函数；
• 回调失败分支使用Continuation.resumeWithException(error)函数；

 4. CoroutineContext 协程上下文包含一个协程调度器。