Kotlin原理

  1. // 普通函数
  2. fun requestToken(): Token {
  3.     // makes request for a token & waits
  4.     return token // returns result when received 
  5. }
  7.  // 挂起函数
  8. suspend fun requestToken(): Token { ... }  
  10. //
  11. fun postItem(item: Item) {
  12.     GlobalScope.launch {
  13.         val token = requestToken()
  14.         val post = createPost(token, item)
  15.         processPost(post)
  16.     }
  17. }

实际上在 JVM 中：

  1. Object requestToken(Continuation<Token> cont) { ... }

Continuation 是什么？

  1. // 像一个回调接口
  2. public interface Continuation<in T> {
  3.     public val context: CoroutineContext
  4.     public fun resumeWith(result: Result<T>)
  5. }

协程内部实现不是使用普通回调的形式，而是使用状态机来处理不同的挂起点

协程的创建与启动

从新建一个协程开始分析协程的创建，
最常见的协程创建方式为CoroutineScope.launch {}，
关键源码如下：

  1. public fun CoroutineScope.launch(
  2.     context: CoroutineContext = EmptyCoroutineContext,
  3.     start: CoroutineStart = CoroutineStart.DEFAULT,
  4.     block: suspend CoroutineScope.() -> Unit
  5. ): Job {
  6.     ...
  7.     coroutine.start(start, coroutine, block) //
  8.     return coroutine
  9. }

coroutine.start(start, coroutine, block)
默认情况下会走到startCoroutineCancellable，
最终会调用到createCoroutineUnintercepted。

  1. internal fun <T> (suspend () -> T).startCoroutineCancellable(completion: Continuation<T>) =
  2.     createCoroutineUnintercepted(completion).intercepted().resumeCancellable(Unit)

public actual fun (suspend () -> T).createCoroutineUnintercepted(
completion: Continuation) : Continuation { … }

  • suspend 函数本身执行需要一个Continuation 实例在恢复时调用，即参数completion
  • 返回值Continuation 是创建出来的协程的载体，receiver suspend 函数会被传给该实例作为协程的实际执行体

在start后，执行到的是 CoroutineStart #invoke()

  1. # CoroutineStart.kt
  3. @InternalCoroutinesApi
  4. public operator fun <T> invoke(block: suspend () -> T, completion: Continuation<T>): Unit =
  5.     when (this) {
  6.         DEFAULT -> block.startCoroutineCancellable(completion)
  7.         ATOMIC ->  block.startCoroutine(completion)
  8.         UNDISPATCHED -> block.startCoroutineUndispatched(completion)
  9.         LAZY -> Unit // will start lazily
  10.     }

这里以DEFAULT为例，startCoroutineCancellable()

  1. internal fun <T> (suspend () -> T).startCoroutineCancellable(completion: Continuation<T>) =
  2.     createCoroutineUnintercepted(completion).intercepted().resumeCancellable(Unit)
  5. public fun <T> Continuation<T>.resumeCancellableWith(
  6.     result: Result<T>,
  7.     onCancellation: ((cause: Throwable) -> Unit)? = null
  8. ): Unit = when (this) {
  9.     is DispatchedContinuation -> resumeCancellableWith(result, onCancellation)
  10.     else -> resumeWith(result) // 传递
  11. }

而在startCoroutineCancellable中，创建了Continuation，且调用resumeWith来传递请求结果。

Continuation是一个接口，接口内有一个resumeWith方法。
在startCoroutineCancellable调用了resumeWith，既然是接口且被调用，那必然是有地方实现了该接口，并且在resumeWith中做了一些事情。

Continuation的具体实现是在ContinuationImpl类中

  1. internal abstract class ContinuationImpl(
  2.     completion: Continuation<Any?>?,
  3.     private val _context: CoroutineContext?
  5. ) : BaseContinuationImpl(completion)


而ContinuationImpl继承自BaseContinuationImpl，
在BaseContinuationImpl中就可以看到resumeWith的具体实现。

  1. internal abstract class BaseContinuationImpl(
  2.     public val completion: Continuation<Any?>?
  3. ) : Continuation<Any?>, CoroutineStackFrame, Serializable {
  5.     public final override fun resumeWith(result: Result<Any?>) {
  6.         var current = this
  7.         var param = result
  8.         while (true) {
  10.             probeCoroutineResumed(current)
  11.             with(current) {
  12.                 val completion = completion!! 
  13.                 val outcome: Result<Any?> =
  14.                     try {
  15.                         val outcome = invokeSuspend(param)   // 判断
  16.                         if (outcome === COROUTINE_SUSPENDED) return
  17.                         Result.success(outcome)
  18.                     } catch (exception: Throwable) {
  19.                         Result.failure(exception)
  20.                     }
  21.                 releaseIntercepted() // this state machine instance is terminating
  22.                 if (completion is BaseContinuationImpl) {
  23.                     // unrolling recursion via loop
  24.                     current = completion
  25.                     param = outcome
  26.                 } else {
  28.                     completion.resumeWith(outcome)
  29.                     return
  30.                 }
  31.             }
  32.         }
  33.     }

可以看到，调用invokeSuspend并且对返回的标志判断

  2. // 举个例子，从编译后的代码中看
  4. suspend fun requestToken(): Token { ... }   // 挂起函数
  5. suspend fun createPost(token: Token, item: Item): Post { ... }  // 挂起函数
  6. fun processPost(post: Post) { ... }
  8. fun postItem(item: Item) {
  9.     GlobalScope.launch {
  10.         val token = requestToken()
  11.         val post = createPost(token, item)
  12.         processPost(post)
  13.     }
  14. }
  18. // 编译后生成的内部类大致如下
  19. final class postItem$1 extends SuspendLambda ... {
  21.     public final Object invokeSuspend(Object result) {   //  invokeSuspend
  22.         ...
  23.         switch (this.label) {
  24.             case 0:
  25.                 this.label = 1;
  26.                 token = requestToken(this)
  27.                 break;
  28.             case 1:
  29.                 this.label = 2;
  30.                 Token token = result;
  31.                 post = createPost(token, this.item, this)
  32.                 break;
  33.             case 2:
  34.                 Post post = result;
  35.                 processPost(post)
  36.                 break;
  37.         }
  38.     }
  39. }

挂起函数会被编译为一个匿名类，这个匿名类中的一个函数实现了这个状态机。
成员变量label代表了当前状态机的状态。
每一个续体(即挂起点中间部分以及挂起点与函数头尾之间的部分)都各自对应了一个状态。
当函数运行到每个挂起点时，lable的值都受限会发生改变，并且当前的续体都会作为实体参数传递给发生了CPS变换的挂起函数。
如果这个挂起函数没有发生事实上的挂起，函数继续运行，如果发生了事实上的挂起，则函数直接return。
由于label记录了状态，所以在协程恢复的时候，可以根据状态使用goto语句直接跳转至上次的挂起点并向后执行，

通过状态机将多个续体融合到一个SuspendLambda对象中，通过维护一个Int类型的状态值，标记续体执行到哪个步骤了，这个状态值和对应的switch语句就是状态机。使用状态机使得无论挂起lambda表达式体内有多少挂起点，编译器也只创建一个SuspendLambda子类和对象。

  1. package kotlin.coroutines
  2. /**续体接口：代表协程下一步应该执行的代码*/
  3. public interface Continuation<in T> {
  4.     public val context: CoroutineContext
  5.     public fun resumeWith(result: Result<T>)
  6. }
  8. package kotlin.coroutines.jvm.internal 
  9. /**②. 续体的基本实现类，实现了resumeWith()*/
  10. internal abstract class BaseContinuationImpl(
  11.         public val completion: Continuation<Any?>?
  12. ) : Continuation<Any?>,... {
  13.     //实现续体的resumeWith()函数，并且是final的子类不能覆盖
  14.     public final override fun resumeWith(result: Result<Any?>) { ... }
  15.     //★定义了一个invokeSuspend()抽象函数，这个函数的函数体实现就是协程代码块中的代码，由kotlin编译器自动生成实现
  16.     protected abstract fun invokeSuspend(result: Result<Any?>): Any?
  17. }
  19. /**③. 续体实现类，继承自BaseContinuationImpl，增加了拦截器intercepted()功能，实现线程调度等*/
  20. internal abstract class ContinuationImpl(
  21.         completion: Continuation<Any?>?,
  22.         private val _context: CoroutineContext?
  23. ) : BaseContinuationImpl(completion) {
  24.       private var intercepted: Continuation<Any?>? = null
  25.      public fun intercepted(): Continuation<Any?> =
  26.         intercepted?: (context[ContinuationInterceptor]?.interceptContinuation(this) ?: this)
  27.                 .also { intercepted = it }
  28. }
  30. /**③. 协程构建器launch{}传入的挂起Lambda表达式的封装抽象类，同时它又是一个续体*/
  31. internal abstract class SuspendLambda(
  32.         public override val arity: Int,
  33.         completion: Continuation<Any?>?
  34. ) : ContinuationImpl(completion), FunctionBase<Any?>, SuspendFunction {}

切换线程

在前面的这段分析中：

  1. internal fun <T> (suspend () -> T).startCoroutineCancellable(completion: Continuation<T>) =
  2.     createCoroutineUnintercepted(completion).intercepted().resumeCancellable(Unit)
  5. public fun <T> Continuation<T>.resumeCancellableWith(
  6.     result: Result<T>,
  7.     onCancellation: ((cause: Throwable) -> Unit)? = null
  8. ): Unit = when (this) {  // 判断
  9.     is DispatchedContinuation -> resumeCancellableWith(result, onCancellation)
  10.     else -> resumeWith(result) // 传递
  11. }

首先会判断当前续体类型是否是DispatchedContinuation（有没有被调度器拦截包装），如果没有被拦截直接调用resumeWith()恢复协程代码块的执行，否则调用拦截包装后的DispatchedContinuation类型的resumeCancellableWith():

  1. //包装后的续体类型，维护了调度器dispatcher和原续体对象continuation
  3. internal class DispatchedContinuation<in T>(
  4.     @JvmField val dispatcher: CoroutineDispatcher,
  5.     @JvmField val continuation: Continuation<T>
  6. ) : DispatchedTask<T>(MODE_UNINITIALIZED), CoroutineStackFrame, Continuation<T> by continuation {
  8.     //DispatchedContinuation类的resumeCancellableWith()
  9.     inline fun resumeCancellableWith(
  10.         result: Result<T>,
  11.         noinline onCancellation: ((cause: Throwable) -> Unit)?
  12.     ) {
  13.         val state = result.toState(onCancellation)
  14.         //通过调度器判断是否需要切换线程
  15.         if (dispatcher.isDispatchNeeded(context)) {
  16.             _state = state
  17.             resumeMode = MODE_CANCELLABLE
  18.             //A 如果需要，则切换线程，将当前续体this作为Runnable传入
  19.             dispatcher.dispatch(context, this)
  20.         } else {
  21.             executeUnconfined(state, MODE_CANCELLABLE) {
  22.                 if (!resumeCancelled(state)) {
  23.                     //★★★B 该函数最终会调用原continuation.resumeWith(result)直接在当前线程恢复协程执行
  24.                     resumeUndispatchedWith(result) 
  25.                 }
  26.             }
  27.         }
  28.     }
  30.     //B 没有调度器拦截的情况直接在当前线程执行
  31.     inline fun resumeUndispatchedWith(result: Result<T>) {
  32.         withCoroutineContext(context, countOrElement) {
  33.             continuation.resumeWith(result)  
  34.         }
  35.     }
  36. }

首先通过调度器判断是否需要切换线程（当前线程是否是调度器使用的线程，如果不是就需要切换），如果需要则将当前续体对象this当作一个Runnable。
(被包装后的DispatchedContinuation续体实现了Runnable)扔给调度器。

  1. //Dispatchers.Default默认调度器的实现类
  2. internal object DefaultScheduler : ExperimentalCoroutineDispatcher() {
  3. }
  4. public open class ExperimentalCoroutineDispatcher(...){
  5.     override val executor: Executor
  6.         get() = coroutineScheduler
  7.     private var coroutineScheduler = createScheduler()
  8.     //A.实现线程切换
  9.     override fun dispatch(context: CoroutineContext, block: Runnable): Unit =
  10.         try {
  11.             //就是将当前续体当作Runnable扔到线程池，coroutineScheduler的类型是Executor
  12.             coroutineScheduler.dispatch(block)
  13.         } catch (e: RejectedExecutionException) {
  14.             DefaultExecutor.dispatch(context, block)
  15.         }
  16. }
  18. //被包装后的DispatchedContinuation续体通过继承DispatchedTask间接实现了Runnable
  19. internal class DispatchedContinuation<in T>(...) : DispatchedTask...{
  20.     override val delegate: Continuation<T>
  21.         get() = this
  22. }
  23. internal abstract class DispatchedTask<in T>(
  24.     @JvmField public var resumeMode: Int
  25. ) : SchedulerTask() {
  26.     internal abstract val delegate: Continuation<T>  //DispatchedContinuation对象
  27.     ...
  28.     //Runnable实现
  29.     public final override fun run() {
  30.         ...
  31.         val delegate = delegate as DispatchedContinuation<T>
  32.         //从代理续体中获取被代理的续体对象，也就是第二次创建的SuspendLambda子类对象
  33.         val continuation = delegate.continuation
  34.         val context = continuation.context
  35.         ...
  36.         withCoroutineContext(context, delegate.countOrElement) {
  37.             ...
  38.             //★★★最终调用原始续体对象的resumeWith()，从而第一次触发invokeSuspend()启动协程
  39.             continuation.resumeWithStackTrace(cause)
  40.             ...
  41.         }
  42.         ...
  43.     }
  44. }
  46. internal inline fun Continuation<*>.resumeWithStackTrace(exception: Throwable) {
  47.     resumeWith(Result.failure(recoverStackTrace(exception, this)))
  48. }

默认的调度器DefaultScheduler中维护了一个线程池，也就是将this扔到线程池中执行从而完成默认调度器的线程切换，DispatchedContinuation的run()方法则从代理续体对象中获取到了原始SuspendLambda子类续体对象，然后调用原始续体的resumeWith()第一次触发invokeSuspend

使用：

密封类

协程密封类解决条件处理

改进前：

  2. interface Result {
  3.     class Success(val msg:String):Result
  4.     class Failure(val error:Exception):Result
  6.     fun getResultMsg(result:Result)=when(result){
  7.         is Success -> result.msg
  8.         is Failure -> result.error.message
  9.         else -> throw IllegalAccessException()
  10.     }
  11. }

存在问题：
（1）else 分支是多余的
（2）else导致潜在分险：当一个类实现Result这个接口后，忘记在getResultMsg（）方法中添加相应条件分支，编译器不会提醒而是当成else 处理

改进后：

  1. sealed class Result
  3. class Success(val msg: String) : Result()
  5. class Failure(val error: Exception) : Result()
  8. fun getResultMsg(result: Result) = when (result) {
  9.     is Success -> result.msg
  10.     is Failure -> "Error is ${result.error.message}"
  11. }

好处： 当when语句中传入一个密封类变量作为条件时，kotlin 编译器会自动检查该密封类有那几个子类，并强制要求将每一个子类所对应的条件全部处理


未改造前

  2. import retrofit2.http.Field
  3. import retrofit2.http.FormUrlEncoded
  4. import retrofit2.http.POST
  6. interface WanAndroidAPI {
  7.     @POST("/user/login")
  8.     @FormUrlEncoded
  9.     fun loginAction(@Field("username")username:String,
  10.                     @Field("password")password:String)
  12.             :retrofit2.Call<LoginRegisterResponseWrapper<LoginRegisterResponse>>
  14.     @POST("/user/register")
  15.     @FormUrlEncoded
  16.     fun registerAction(@Field("usename")username: String,
  17.                        @Field("password")password: String,
  18.                        @Field("repassword")repassword:String)
  19.             :retrofit2.Call<LoginRegisterResponseWrapper<LoginRegisterResponse>>
  20.    //------------------
  24. }
  1. //LoginRegisterResponse.kt
  3. data class LoginRegisterResponse(val admin:Boolean,
  4.                                  val chapterTops:List<*>,
  5.                                  val collectIds:List<*>,
  6.                                  val email:String?,
  7.                                  val icon:String?,
  8.                                  val id:String?,
  9.                                  val nickname:String?,
  10.                                  val password:String?,
  11.                                  val publicName:String?,
  12.                                  val token:String?,
  13.                                  val type:Int,
  14.                                  val username:String?
  15. )
  1. //APIClient.kt
  3. import okhttp3.OkHttpClient
  4. import retrofit2.Retrofit
  5. import java.util.concurrent.TimeUnit
  7. class APIClient {
  8.     private object Holder{
  9.         val INSTANCE = APIClient()
  11.     }
  13.     companion object{
  14.         val instance=Holder.INSTANCE
  15.     }
  17.     fun<T> instanceRetrofit(apiInterface:Class<T>):T{
  18.         val mOkHttpClient = OkHttpClient().newBuilder().myApply {
  19.                 readTimeout(10000,TimeUnit.SECONDS)
  20.                 connectTimeout(1000,TimeUnit.SECONDS)
  21.                 writeTimeout(1000,TimeUnit.SECONDS)
  22.             }.build()
  24.         val retrofit:Retrofit = Retrofit.Builder()
  25.             .baseUrl("https://www.wanandroid.com")
  26.             .client(mOkHttpClient)
  27.             .addCallAdapterFactory(RxJava2CallAdapterFactory.create()) 
  28.             .addConverterFactory(GsonConverterFactory.create())  
  29.             .build()
  31.             return retrofit.create(apiInterface)
  33.     }
  35. }
  37. fun <T> T.myApply(mm:T.()->Unit):T{
  38.     mm()
  39.     return this
  40. }
  1. //LoginRegisterResponseWrapper.kt
  2. data class LoginRegisterResponseWrapper<T> (val data:T,val errorCode:Int,val errorMsg:String)
  1. //LoginRegisterResponse.kt
  2. data class LoginRegisterResponse(val admin:Boolean,
  3.                                  val chapterTops:List<*>,
  4.                                  val collectIds:List<*>,
  5.                                  val email:String?,
  6.                                  val icon:String?,
  7.                                  val id:String?,
  8.                                  val nickname:String?,
  9.                                  val password:String?,
  10.                                  val publicName:String?,
  11.                                  val token:String?,
  12.                                  val type:Int,
  13.                                  val username:String?
  14. )
  1. //MainActivity.kt
  3. import android.app.ProgressDialog
  4. import androidx.appcompat.app.AppCompatActivity
  5. import android.os.Bundle
  6. import android.os.Handler
  7. import android.os.Looper
  8. import android.util.Log
  9. import android.view.View
  10. import android.widget.TextView
  13. class MainActivity : AppCompatActivity() {
  15.     private val TAG = "ckb"
  16.     var mProgressDialog:ProgressDialog?=null
  17.     var textView:TextView?=null
  19.     //第二大步：主自线程，更新UI
  20.     val mHandler = Handler(Looper.getMainLooper()){
  21.         val result = it.obj as LoginRegisterResponseWrapper<LoginRegisterResponse>
  22.         Log.d(TAG,"errorMsg:${result.data}")
  23.         textView?.setText(result.data.toString())
  24.         mProgressDialog?.dismiss()
  25.         false
  26.     }
  29.     //
  31.     override fun onCreate(savedInstanceState: Bundle?) {
  32.         super.onCreate(savedInstanceState)
  33.         setContentView(R.layout.activity_main)
  34.         textView = findViewById<TextView>(R.id.textView)
  35.     }
  37.     fun startRequest(view:View){
  38.         mProgressDialog = ProgressDialog(this)
  39.         mProgressDialog ?.setTitle("请求服务器中")
  40.         mProgressDialog?.show()
  42.         //开启一步线程
  43.         object:Thread(){
  44.             override fun run() {
  45.                 super.run()
  46.                 Thread.sleep(2000)
  47.                 val loginResult =
  48.                     APIClient.instance.instanceRetrofit(WanAndroidAPI::class.java)
  49.                         .loginAction("Derry-vip","123456")
  51.                 val result:LoginRegisterResponseWrapper<LoginRegisterResponse>?=loginResult.execute().body()
  53.                 //发送Handler 切换Android 主线程
  54.                 val msg =mHandler.obtainMessage()
  55.                 msg.obj =result
  56.                 mHandler.sendMessage(msg)
  57.             }
  58.         }.start()
  59.     }
  62. }
  1. <?xml version="1.0" encoding="utf-8"?>
  2. <LinearLayout xmlns:android="http://schemas.android.com/apk/res/android"
  3.     xmlns:app="http://schemas.android.com/apk/res-auto"
  4.     xmlns:tools="http://schemas.android.com/tools"
  5.     android:layout_width="match_parent"
  6.     android:layout_height="match_parent"
  7.     tools:context=".MainActivity">
  9.     <TextView
  10.         android:id="@+id/textView"
  11.         android:layout_width="wrap_content"
  12.         android:layout_height="wrap_content"
  13.         android:text="Hello World!"
  14.         app:layout_constraintBottom_toBottomOf="parent"
  15.         app:layout_constraintLeft_toLeftOf="parent"
  16.         app:layout_constraintRight_toRightOf="parent"
  17.         app:layout_constraintTop_toTopOf="parent" />
  19.     <Button
  20.         android:layout_marginTop="400dp"
  21.         android:text="网络请求"
  22.         android:layout_width="wrap_content"
  23.         android:layout_height="wrap_content"
  24.         android:onClick="startRequest"
  25.         />
  27. </LinearLayout>

改用协程解决Hander

  2. // WanAndroidAPI.kt
  4. import retrofit2.http.Field
  5. import retrofit2.http.FormUrlEncoded
  6. import retrofit2.http.POST
  8. interface WanAndroidAPI {
  9.     @POST("/user/login")
  10.     @FormUrlEncoded
  11.     fun loginAction(@Field("username")username:String,
  12.                     @Field("password")password:String)
  14.             :retrofit2.Call<LoginRegisterResponseWrapper<LoginRegisterResponse>>
  16.     @POST("/user/register")
  17.     @FormUrlEncoded
  18.     fun registerAction(@Field("usename")username: String,
  19.                        @Field("password")password: String,
  20.                        @Field("repassword")repassword:String)
  21.             :retrofit2.Call<LoginRegisterResponseWrapper<LoginRegisterResponse>>
  22.    //--------   ----------
  24.     // 协程写法-----加上suspend, 同时不再需要Call<>
  25.     @POST("/user/login")
  26.     @FormUrlEncoded
  27.     suspend fun loginActionCoroutine(@Field("username")username:String,
  28.                     @Field("password")password:String)
  30.             :LoginRegisterResponseWrapper<LoginRegisterResponse>
  32.     @POST("/user/register")
  33.     @FormUrlEncoded
  34.     fun registerActionCoroutine(@Field("usename")username: String,
  35.                        @Field("password")password: String,
  36.                        @Field("repassword")repassword:String)
  37.             :LoginRegisterResponseWrapper<LoginRegisterResponse>
  40. }
  1. //
  3. class MainActivity : AppCompatActivity() {
  5.     private val TAG = "ckb"
  6.     var mProgressDialog:ProgressDialog?=null
  7.     var textView:TextView?=null
  9. //    //第二大步：主自线程，更新UI
  10. //    val mHandler = Handler(Looper.getMainLooper()){
  11. //        val result = it.obj as LoginRegisterResponseWrapper<LoginRegisterResponse>
  12. //        Log.d(TAG,"errorMsg:${result.data}")
  13. //        textView?.setText(result.data.toString())
  14. //        mProgressDialog?.dismiss()
  15. //        false
  16. //    }
  19.     //
  21.     override fun onCreate(savedInstanceState: Bundle?) {
  22.         super.onCreate(savedInstanceState)
  23.         setContentView(R.layout.activity_main)
  24.         textView = findViewById<TextView>(R.id.textView)
  25.     }
  27.     fun startRequest(view:View){
  28.         mProgressDialog = ProgressDialog(this)
  29.         mProgressDialog ?.setTitle("请求服务器中")
  30.         mProgressDialog?.show()
  32. //        //开启一步线程
  33. //        object:Thread(){
  34. //            override fun run() {
  35. //                super.run()
  36. //                Thread.sleep(2000)
  37. //                //---------
  38. //                val loginResult =
  39. //                    APIClient.instance.instanceRetrofit(WanAndroidAPI::class.java)
  40. //                        .loginAction("Derry-vip","123456")
  41. //                //--------
  42. //
  43. //                val result:LoginRegisterResponseWrapper<LoginRegisterResponse>?=loginResult.execute().body()
  44. //
  45. //                //发送Handler 切换Android 主线程
  46. //                val msg =mHandler.obtainMessage()
  47. //                msg.obj =result
  48. //                mHandler.sendMessage(msg)
  49. //            }
  50. //        }.start()
  53.         // -----------------------协程改写-----------
  55.         GlobalScope.launch(Dispatchers.Main) {//在Android 上，默认是异步处理
  56.             // 主动会被挂起
  57.             // 内部会切换成异步线程请求服务器
  59.             val loginResult =
  60.                 APIClient.instance.instanceRetrofit(WanAndroidAPI::class.java)
  61.                     .loginActionCoroutine("Derry-vip","123456")
  63.             // 执行完成后，主动切换回安卓主线程
  64.             //在主线程更新UI
  65.             Log.d(TAG,"errorMsg:${loginResult.data}")
  66.             textView?.setText(loginResult.data.toString())
  67.             mProgressDialog?.dismiss()
  68.         }
  70.       //---------------------------------
  73.     }
  75. }

协程解决多层回调带来的问题

  1. // MainActivity3.kt
  3. import android.app.ProgressDialog
  4. import android.graphics.Color
  5. import android.os.Bundle
  6. import android.os.Handler
  7. import android.os.Looper
  8. import android.os.Message
  9. import android.view.View
  10. import android.widget.TextView
  11. import androidx.appcompat.app.AppCompatActivity
  13. // 接口
  14. interface ResponseCallback{
  16.     fun responseSuccess(serverResponseInfo:String)// 登陆成功后的信息类
  17.     fun responseError(serverResponseErrorMsg:String) // 登陆失败后的描述
  19. }
  20. //请求加载用户的数据
  21. private fun requestLoadUser(responseCallback:ResponseCallback){
  22.     val isLoadSuccess = true;
  23.     //开启异步线程
  24.     object:Thread(){
  25.         override fun run(){
  26.             super.run()
  27.             try{
  28.                 sleep(3000L)
  29.                 if (isLoadSuccess){
  30.                     responseCallback.responseSuccess("加载到【用户数据】信息集")
  31.                 }else {
  32.                     responseCallback.responseError("加载【用户数据】，加载失败，服务器宕机")
  33.                 }
  34.             } catch(e:InterruptedException){
  35.                 e.printStackTrace()
  36.             }
  37.         }
  38.     }.start()
  40. }
  42. //
  43. //请求加载[用户资产数据]
  44. private fun requestLoadUserAssets(responseCallback:ResponseCallback){
  45.     val isLoadSuccess = true;
  46.     //开启异步线程
  47.     object:Thread(){
  48.         override fun run(){
  49.             super.run()
  50.             try{
  51.                 sleep(3000L)
  52.                 if (isLoadSuccess){
  53.                     responseCallback.responseSuccess("加载到【用户资产数据】信息集")
  54.                 }else {
  55.                     responseCallback.responseError("加载【用户资产数据】，加载失败，服务器宕机")
  56.                 }
  57.             }catch(e:InterruptedException){
  58.                 e.printStackTrace()
  59.             }
  60.         }
  61.     }.start()
  63. }
  64. //
  65. //请求加载[用户资产详情数据】
  66. private fun requestLoadUserAssetsDetails(responseCallback:ResponseCallback){
  67.     val isLoadSuccess = true;
  68.     //开启异步线程
  69.     object:Thread(){
  70.         override fun run(){
  71.             super.run()
  72.             try{
  73.                 sleep(3000L)
  74.                 if (isLoadSuccess){
  75.                     responseCallback.responseSuccess("加载到【用户资产详情数据】信息集")
  76.                 }else {
  77.                     responseCallback.responseError("加载【用户资产详情数据】，加载失败，服务器宕机")
  78.                 }
  79.             }catch(e:InterruptedException){
  80.                 e.printStackTrace()
  81.             }
  82.         }
  83.     }.start()
  85. }
  89. class MainActivity3 :AppCompatActivity(){
  91.     private val TAG = "ckb"
  92.     var mProgressDialog: ProgressDialog?=null
  93.     var textView: TextView?=null
  95.     override fun onCreate(savedInstanceState: Bundle?) {
  96.         super.onCreate(savedInstanceState)
  97.         setContentView(R.layout.activity_main)
  98.         textView = findViewById<TextView>(R.id.textView)
  99.     }
  101.     fun startRequest(view: View) {
  102.         mProgressDialog = ProgressDialog(this)
  103.         mProgressDialog?.setTitle("请求服务器中")
  104.         mProgressDialog?.show()
  106.         // 先进行第一步异步请求
  107.         requestLoadUser(object:ResponseCallback{
  108.             override fun responseSuccess(serverResponseInfo: String) {
  109.                 //从异步线程切回主线程
  110.                 val handler:Handler = object :Handler(Looper.getMainLooper()){
  111.                     override fun handleMessage(msg: Message) {
  112.                         super.handleMessage(msg)
  113.                         //第一次更新UI
  114.                         textView?.text = serverResponseInfo
  115.                         textView?.setTextColor(Color.RED)
  117.                         //---第二次请求--
  118.                         requestLoadUserAssets(object : ResponseCallback {
  119.                             override fun responseSuccess(serverResponseInfo: String) {
  120.                                 val handler:Handler = object :Handler(Looper.getMainLooper()) {
  121.                                     override fun handleMessage(msg: Message) {
  122.                                         super.handleMessage(msg)
  123.                                         //第二次更新UI
  124.                                         textView?.text = serverResponseInfo
  125.                                         textView?.setTextColor(Color.RED)
  127.                                         //----第三次请求------
  128.                                         requestLoadUserAssetsDetails(object:ResponseCallback{
  129.                                             override fun responseSuccess(serverResponseInfo: String) {
  130.                                                 val handler:Handler = object :Handler(Looper.getMainLooper()) {
  131.                                                     override fun handleMessage(msg: Message) {
  132.                                                         super.handleMessage(msg)
  133.                                                         //第三次更新UI
  134.                                                         textView?.text = serverResponseInfo
  135.                                                         mProgressDialog?.dismiss()
  136.                                                         textView?.setTextColor(Color.BLACK)
  137.                                                     }
  138.                                                 }
  139.                                                 handler.sendEmptyMessage(0)
  140.                                             }
  142.                                             override fun responseError(serverResponseErrorMsg: String) {
  143.                                                 TODO("Not yet implemented")
  144.                                             }
  146.                                         })
  147.                                     }
  148.                                 }
  149.                                 handler.sendEmptyMessage(0)
  150.                             }
  152.                             override fun responseError(serverResponseErrorMsg: String) {
  153.                                 TODO("Not yet implemented")
  154.                             }
  156.                         })
  158.                         //----
  159.                     }
  160.                 }
  161.                 handler.sendEmptyMessage(0)
  162.             }
  164.             override fun responseError(serverResponseErrorMsg: String) {
  165.                 TODO("Not yet implemented")
  166.             }
  168.         })
  169.     }
  175. }

协程更改后

  1. import android.app.ProgressDialog
  2. import android.graphics.Color
  3. import android.os.Bundle
  4. import android.view.View
  5. import android.widget.TextView
  6. import androidx.appcompat.app.AppCompatActivity
  7. import kotlinx.coroutines.*
  9. //
  10. //请求加载用户的数据
  11. private suspend fun requestLoadUser():String{
  12.     val isLoadSuccess = true;
  13.     // 用withContext 手动切换到异步线程
  14.     withContext(Dispatchers.IO){
  15.         delay(3000L) // 模拟耗时
  16.     }
  17.     if(isLoadSuccess){
  18.         return "加载到【用户数据】信息集"
  19.     }else{
  20.         return "加载【用户数据】加载失败，服务器宕机"
  21.     }
  23. }
  24. //请求加载【用户资产】的数据
  25. private suspend fun requestLoadUserAssets():String{
  26.     val isLoadSuccess = true;
  27.     // 用withContext 手动切换到异步线程
  28.     withContext(Dispatchers.IO){
  29.         delay(3000L) // 模拟耗时
  30.     }
  31.     if(isLoadSuccess){
  32.         return "加载到【用户资产】信息集"
  33.     }else{
  34.         return "加载【用户数据资产】加载失败，服务器宕机"
  35.     }
  37. }
  38. //请求加载[用户资产详情数据】
  39. private suspend fun requestLoadUserAssetsDetails():String{
  40.     val isLoadSuccess = true;
  41.     // 用withContext 手动切换到异步线程
  42.     withContext(Dispatchers.IO){
  43.         delay(3000L) // 模拟耗时
  44.     }
  45.     if(isLoadSuccess){
  46.         return "加载到【用户资产详情数据】信息集"
  47.     }else{
  48.         return "加载【用户资产详情数据】加载失败，服务器宕机"
  49.     }
  51. }
  56. class MainActivity4 :AppCompatActivity() {
  57.     private val TAG = "ckb"
  58.     var mProgressDialog: ProgressDialog?=null
  59.     var textView: TextView?=null
  61.     override fun onCreate(savedInstanceState: Bundle?) {
  62.         super.onCreate(savedInstanceState)
  63.         setContentView(R.layout.activity_main)
  64.         textView = findViewById<TextView>(R.id.textView)
  65.     }
  67.     fun startRequest(view: View){
  68.         mProgressDialog= ProgressDialog(this)
  69.         mProgressDialog?.setTitle("请求服务器中。。。")
  70.         mProgressDialog?.show()
  72.         //这里缺少了错误处理的逻辑
  73.         GlobalScope.launch ( Dispatchers.Main ){
  74.             // 异步请求1
  75.             var serverResponseInfo = requestLoadUser() // 这一步自动完成了主线程和异步线程的切换
  76.             //拿到数据后立刻更新UI
  77.             textView?.text = serverResponseInfo
  78.             textView?.setTextColor(Color.RED)
  80.             //执行异步操作2
  81.             serverResponseInfo = requestLoadUserAssets()
  82.             //拿到数据后立刻更新UI
  83.             textView?.text = serverResponseInfo
  84.             textView?.setTextColor(Color.BLUE)
  86.             //执行异步操作3
  87.             serverResponseInfo = requestLoadUserAssets()
  88.             //拿到数据后立刻更新UI
  89.             textView?.text = serverResponseInfo
  90.             mProgressDialog?.dismiss()
  91.             textView?.setTextColor(Color.RED)
  93.         }
  94.     }
  95. }

Kotlin 协程：使用async和await实现高效并发

  1. //计算一
  2. private suspend fun intValue1(): Int {    
  3.     delay(1000)
  4.     return 1
  5. }
  7.  // 计算二
  8. private suspend fun intValue2(): Int { 
  9.     delay(2000)
  10.     return 2
  11. }
  15. fun main() = runBlocking {
  16.     val elapsedTime = measureTimeMillis {
  17.         val value1 = intValue1()
  18.         val value2 = intValue2()
  19.         println("the result is ${value1 + value2}")
  20.     }
  22.     println("the elapsedTime is $elapsedTime")
  23. }
  25. // 综合
  26. public inline fun measureTimeMillis(block: () -> Unit): Long {
  27.     val start = System.currentTimeMillis()
  28.     block()
  29.     return System.currentTimeMillis() - start
  30. }

intValue1()和intValue2(）是完全独立的，但是从运行结果来看，耗费的时间是intValue1()的时间 + intValue2()的时间。

使用async和await改进：

  1. fun main() = runBlocking {
  3.     val elapsedTime = measureTimeMillis {
  4.         val value1 = async { intValue1() }
  5.         val value2 = async { intValue2() }
  7.         println("the result is ${value1.await() + value2.await()}")
  8.     }
  10.     println("the elapsedTime is $elapsedTime")
  11. }
  13. private suspend fun intValue1(): Int {
  14.     delay(1000)
  15.     return 1
  16. }
  18. private suspend fun intValue2(): Int {
  19.     delay(2000)
  20.     return 2
  21. }
  1. public fun <T> CoroutineScope.async(
  2.     context: CoroutineContext = EmptyCoroutineContext,
  3.     start: CoroutineStart = CoroutineStart.DEFAULT,
  4.     block: suspend CoroutineScope.() -> T
  5. ): Deferred<T>
  • 它和CoroutineScope.launch{}类似，可以用来创建一个协程，不同的是launch的返回结果是Job类型，而aysnc的返回结果是Deferred类型。从类的层次结构上看，Deferred是Job的子接口；从功能上来看，Deferred就是带返回结果的Job。
  • CoroutineStart.LAZY的作用，可以使得async启动的两个协程并没有立即执行。而是直到调用await方法之后，才开始执行，而await又是会去等待结果，自然需要等待intValue1协程执行完毕后，遇到intValue2.await()，才会触发intValue2协程的执行，又要去等待。那么，async的并发也就失效了。

https://blog.csdn.net/xlh1191860939/article/details/104981066

Coroutines Flow

作用：异步返回多个计算好的值呢

Flow是一种类似于序列的冷流，flow构建器中的代码直到流被收集的时候才运行

  1. fun simpleFlow() = flow {
  2.     println("Flow started")
  3.     for(i in 1..3){
  4.         delay(1000)
  5.         emit(i)
  6.     }
  7. }
  9. fun testFlowIsCode() = runBlocking {
  10.     val flow = simpleFlow()
  11.     println("Flow Collect")
  12.     flow.collect { println(it) }
  13.     println("Flow Collect again")
  14.     flow.collect { println(it) }
  15. }
  1. Flow Collect
  2. Flow started
  3. 1
  4. 2
  5. 3
  6. Flow Collect again
  7. Flow started
  8. 1
  9. 2
  10. 3
  12. Process finished with exit code 0

代码执行val flow = simpleFlow()的时候没有执行flow{…}构建块中的代码，只有调用collect的时候才执行，这就是冷流

使用 flowOn 切换线程

使用withTimeoutOrNull() 流的取消挂起

Flow 的 Terminal 运算符可以是 suspend 函数，如 collect、single、reduce、toList 等；也可以是 launchIn 运算符，用于在指定 CoroutineScope 内使用 flow。
Flow 在使用各个 suspend 函数时不会阻塞主线程的运行。

  1. //使用 flow：
  3. fun main() = runBlocking {
  5.     launch {
  6.         for (j in 1..5) {
  7.             delay(100)
  8.             println("I'm not blocked $j")
  9.         }
  10.     }
  12.     flow {
  13.         for (i in 1..5) {
  14.             delay(100)
  15.             emit(i)
  16.         }
  17.     }.collect { println(it) }
  19.     println("Done")
  20. }
  23. //-------------结果-----------
  24. 1
  25. I'm not blocked 1
  26. 2
  27. I'm not blocked 2
  28. 3
  29. I'm not blocked 3
  30. 4
  31. I'm not blocked 4
  32. 5
  33. Done
  34. I'm not blocked 5
  1. //使用 sequence：
  3. fun main() = runBlocking {
  5.     launch {
  6.         for (k in 1..5) {
  7.             delay(100)
  8.             println("I'm blocked $k")
  9.         }
  10.     }
  12.     sequence {
  13.         for (i in 1..5) {
  14.             Thread.sleep(100)
  15.             yield(i)
  16.         }
  17.     }.forEach { println(it) }
  19.     println("Done")
  20. }