協程及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 協程上下文包含一個協程調度器。