协程及Kotlin协程

Posted 2020-12-12 naray

一、协程是什么？

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

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

二、协程作用？

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

三、协程分类

• 按调用栈分类：

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

• 按调用关系分类：

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

 

四、Kotlin中协程

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

1. 挂起函数及suspend修饰符

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

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

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

Continuation源码：

  1 /**
  2  * Interface representing a continuation after a suspension point that returns a value of type `T`.
  3  */
  4 @SinceKotlin("1.3")
  5 public interface Continuation<in T> {
  6     /**
  7      * The context of the coroutine that corresponds to this continuation.
  8      */
  9     public val context: CoroutineContext
 10 
 11     /**
 12      * Resumes the execution of the corresponding coroutine passing a successful or failed [result] as the
 13      * return value of the last suspension point.
 14      */
 15     public fun resumeWith(result: Result<T>)
 16 }
 17 
 18 /**
 19  * Classes and interfaces marked with this annotation are restricted when used as receivers for extension
 20  * `suspend` functions. These `suspend` extensions can only invoke other member or extension `suspend` functions on this particular
 21  * receiver and are restricted from calling arbitrary suspension functions.
 22  */
 23 @SinceKotlin("1.3")
 24 @Target(AnnotationTarget.CLASS)
 25 @Retention(AnnotationRetention.BINARY)
 26 public annotation class RestrictsSuspension
 27 
 28 /**
 29  * Resumes the execution of the corresponding coroutine passing [value] as the return value of the last suspension point.
 30  */
 31 @SinceKotlin("1.3")
 32 @InlineOnly
 33 public inline fun <T> Continuation<T>.resume(value: T): Unit =
 34     resumeWith(Result.success(value))
 35 
 36 /**
 37  * Resumes the execution of the corresponding coroutine so that the [exception] is re-thrown right after the
 38  * last suspension point.
 39  */
 40 @SinceKotlin("1.3")
 41 @InlineOnly
 42 public inline fun <T> Continuation<T>.resumeWithException(exception: Throwable): Unit =
 43     resumeWith(Result.failure(exception))
 44 
 45 
 46 /**
 47  * Creates a [Continuation] instance with the given [context] and implementation of [resumeWith] method.
 48  */
 49 @SinceKotlin("1.3")
 50 @InlineOnly
 51 public inline fun <T> Continuation(
 52     context: CoroutineContext,
 53     crossinline resumeWith: (Result<T>) -> Unit
 54 ): Continuation<T> =
 55     object : Continuation<T> {
 56         override val context: CoroutineContext
 57             get() = context
 58 
 59         override fun resumeWith(result: Result<T>) =
 60             resumeWith(result)
 61     }
 62 
 63 /**
 64  * Creates a coroutine without a receiver and with result type [T].
 65  * This function creates a new, fresh instance of suspendable computation every time it is invoked.
 66  *
 67  * To start executing the created coroutine, invoke `resume(Unit)` on the returned [Continuation] instance.
 68  * The [completion] continuation is invoked when the coroutine completes with a result or an exception.
 69  * Subsequent invocation of any resume function on the resulting continuation will produce an [IllegalStateException].
 70  */
 71 @SinceKotlin("1.3")
 72 @Suppress("UNCHECKED_CAST")
 73 public fun <T> (suspend () -> T).createCoroutine(
 74     completion: Continuation<T>
 75 ): Continuation<Unit> =
 76     SafeContinuation(createCoroutineUnintercepted(completion).intercepted(), COROUTINE_SUSPENDED)
 77 
 78 /**
 79  * Creates a coroutine with receiver type [R] and result type [T].
 80  * This function creates a new, fresh instance of suspendable computation every time it is invoked.
 81  *
 82  * To start executing the created coroutine, invoke `resume(Unit)` on the returned [Continuation] instance.
 83  * The [completion] continuation is invoked when the coroutine completes with a result or an exception.
 84  * Subsequent invocation of any resume function on the resulting continuation will produce an [IllegalStateException].
 85  */
 86 @SinceKotlin("1.3")
 87 @Suppress("UNCHECKED_CAST")
 88 public fun <R, T> (suspend R.() -> T).createCoroutine(
 89     receiver: R,
 90     completion: Continuation<T>
 91 ): Continuation<Unit> =
 92     SafeContinuation(createCoroutineUnintercepted(receiver, completion).intercepted(), COROUTINE_SUSPENDED)
 93 
 94 /**
 95  * Starts a coroutine without a receiver and with result type [T].
 96  * This function creates and starts a new, fresh instance of suspendable computation every time it is invoked.
 97  * The [completion] continuation is invoked when the coroutine completes with a result or an exception.
 98  */
 99 @SinceKotlin("1.3")
100 @Suppress("UNCHECKED_CAST")
101 public fun <T> (suspend () -> T).startCoroutine(
102     completion: Continuation<T>
103 ) {
104     createCoroutineUnintercepted(completion).intercepted().resume(Unit)
105 }
106 
107 /**
108  * Starts a coroutine with receiver type [R] and result type [T].
109  * This function creates and starts a new, fresh instance of suspendable computation every time it is invoked.
110  * The [completion] continuation is invoked when the coroutine completes with a result or an exception.
111  */
112 @SinceKotlin("1.3")
113 @Suppress("UNCHECKED_CAST")
114 public fun <R, T> (suspend R.() -> T).startCoroutine(
115     receiver: R,
116     completion: Continuation<T>
117 ) {
118     createCoroutineUnintercepted(receiver, completion).intercepted().resume(Unit)
119 }
120 
121 /**
122  * Obtains the current continuation instance inside suspend functions and suspends
123  * the currently running coroutine.
124  *
125  * In this function both [Continuation.resume] and [Continuation.resumeWithException] can be used either synchronously in
126  * the same stack-frame where the suspension function is run or asynchronously later in the same thread or
127  * from a different thread of execution. Subsequent invocation of any resume function will produce an [IllegalStateException].
128  */
129 @SinceKotlin("1.3")
130 @InlineOnly
131 public suspend inline fun <T> suspendCoroutine(crossinline block: (Continuation<T>) -> Unit): T =
132     suspendCoroutineUninterceptedOrReturn { c: Continuation<T> ->
133         val safe = SafeContinuation(c.intercepted())
134         block(safe)
135         safe.getOrThrow()
136     }
137 
138 /**
139  * Returns the context of the current coroutine.
140  */
141 @SinceKotlin("1.3")
142 @Suppress("WRONG_MODIFIER_TARGET")
143 @InlineOnly
144 public suspend inline val coroutineContext: CoroutineContext
145     get() {
146         throw NotImplementedError("Implemented as intrinsic")
147     }

 挂起函数格式：

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 协程上下文包含一个协程调度器。