1.协程与普通方法任务调度对比
1.执行串行任务
1.普通方式执行
在子线程中执行耗时操作后，通过接口回调来回调结果，当多个任务串行依赖的时候，就会出现 “回调地狱”
object Normal{
private const val TAG = "Normal"
fun startSerialTask(){
request1 { result1 ->
request2(result1,callback = { result2->
request3(result2,callback = { result3->
Log.e(TAG,"get result = $result3 ")
//抛到主线程中更新 UI
HiExecutor.executeOnMain(Runnable {
Log.e(TAG,"update ui para = $result3 thread = ${Thread.currentThread().name}")
})
})
})
}
Log.e(TAG,"startSerialTask ...")
}
/**
* 在子线程执行任务1,拿到结果
*/
fun request1( callback:(result:String)->Unit){
HiExecutor.execute(runnable = Runnable {
Log.e(TAG,"request1 start...")
Thread.sleep(1000)
Log.e(TAG,"request1 end...")
callback.invoke("request1")
})
}
fun request2(paras: String, callback:(result:String)->Unit){
HiExecutor.execute(runnable = Runnable {
Log.e(TAG,"request2 start...")
Thread.sleep(2000)
Log.e(TAG,"request2 end...")
callback.invoke("request2 + $paras")
})
}
fun request3(paras: String, callback:(result:String)->Unit){
HiExecutor.execute(runnable = Runnable {
Log.e(TAG,"request3 start...")
Thread.sleep(2000)
Log.e(TAG,"request3 end...")
callback.invoke("request3 + $paras")
})
}
}
日志结果：
2.协程的方式执行
1.场景描述：现在有三个运行在子线程任务依赖关系：a --> b --> c，执行完后更新 UI
/**
* author : shengping.tian
* time : 2021/09/07
* desc : 在子线程中执行 a->b->c 三个任务后，再在子线程中执行更新 UI 的操作
* version: 1.0
*/
object CoroutineSample1 {
private const val TAG = "CoroutineSample1"
fun startSerialScene() {
GlobalScope.launch(Dispatchers.IO) {
//2.协程启动
Log.e(TAG, "coroutine is running thread = ${Thread.currentThread().name}")
val request1 = request1()
val request2 = request2(request1)
val request3 = request3(request2)
//抛到主线程去更新 UI
Log.e(TAG, "coroutine is end , result = $request3")
updateUI(request3)
}
//1.此处的日志优先于 2 处
Log.e(TAG, "coroutine has launched ")
}
/**
* 1.直接通过协程更新 UI，内部也是通过 handler.post() 方法执行斜程代码块中的方法，
*
* 2.或者直接通过handler将结果抛到主线程
*/
fun updateUI(paras: String){
GlobalScope.launch(Dispatchers.Main) {
Log.e(TAG,"updateUI paras = $paras -- thread = ${Thread.currentThread().name}")
}
}
suspend fun request1(): String {
val threadName = Thread.currentThread().name
Log.e(TAG, "request1 start...$threadName")
delay(1000)
Log.e(TAG, "request1 end...")
return "request1"
}
suspend fun request2(paras: String): String {
val threadName = Thread.currentThread().name
Log.e(TAG, "request2 start... $threadName")
delay(2000)
Log.e(TAG, "request2 end...")
return "request2 + $paras"
}
suspend fun request3(paras: String): String {
val threadName = Thread.currentThread().name
Log.e(TAG, "request3 start... $threadName ")
delay(3000)
Log.e(TAG, "request3 end... ")
return "request3 + $paras"
}
}
执行结果日志：
2.直接在 UI 线程执行三个串行任务后更新 UI
/**
* 直接在 UI 线程执行三个串行的任务 a -> b-> c
*/
fun startSerialScene2() {
GlobalScope.launch(Dispatchers.Main) {
Log.e(TAG, "coroutine is running thread = ${Thread.currentThread().name}")
val request1 = request1()
val request2 = request2(request1)
val request3 = request3(request2)
//当前线程为主线程，为何不会阻塞 UI 线程呢，launch代码块内部最终还是通过 handle.post方法来执行方法，至于任务的挂起和恢复后面再分析。
Log.e(TAG, "coroutine is end , result = $request3")
}
//1.
Log.e(TAG, "coroutine has launched ")
}
日志结果：
2.并行依赖任务
场景描述：在子线程中 request1() --》并行请求 request1() 和 request2() 的结果，最终执行更行 UI 的操作
1.普通的方法执行
普通方法进行多线程的并发控制，可以借助并发工具类 CountDownLatch 来控制，以拿到多个线程的并发结果之后更新UI
object NormalSample2 {
private const val TAG = "NormalSample2"
val countDownLatch = CountDownLatch(2)
fun startAsyncTask() {
Log.e(TAG,"startAsyncTask init... ")
request1 { result1 ->
var result2 = ""
var result3 = ""
request2(result1,callback = { request2->
result2 = request2
countDownLatch.countDown()
})
request3(result1,callback = { request3->
result3 += request3
countDownLatch.countDown()
})
countDownLatch.await()
Log.e(TAG,"startAsyncTask end... ")
updateUI(result2,result3)
}
Log.e(TAG,"startAsyncTask start... ")
}
.....
}
日志结果：
2.协程执行
object CoroutineSample2 {
private const val TAG = "CoroutineSample2"
/**
* 启动一个线程,先执行request1，完了之后，同时运行request2 和request3, 这俩并发都结束了才执行updateIU
*/
fun startAsyncScene2() {
GlobalScope.launch(Dispatchers.IO) {
Log.e(TAG, "coroutine is running")
val result1 = request1()
val deferred2 = GlobalScope.async { request2(result1) }
val deferred3 = GlobalScope.async { request3(result1) }
//不能串行调用，否则 deferred2.await() 会阻塞 deferred3.await() 的执行，一定要一起调用
// deferred2.await();
//deferred3.await()
updateUI(deferred2.await(), deferred3.await())
}
Log.e(TAG, "coroutine has started")
}
.....
}
日志结果：
从 logcat 日志可以看到：request2() 和 request3() 同时执行，并且都执行完之后才执行 updateUI 方法。
2.协程挂机与恢复的原理
1.反编译 suspend 方法源码分析
协程的挂起本质上是方法的挂机，协程的恢复本质上也是方法的恢复。
? 但是为什么当前方法已经调用了，却还能被挂起，然后又能自动恢复呢？普通同步方法一旦执行完 return 之后，继续向下执行了，而异步方法则需要通过 callback 来回调异步方法的执行结果。
? 在 kotlin 中写挂起函数的时候并没有传入任何 callback 接口，它是如何使得异步的方法以同步的方式拿到返回结果的呢？
object CoroutineScene2 {
private val TAG: String = "CoroutineScene2"
suspend fun request1(): String {
val request2 = request2()
return "result from request1 $request2";
}
suspend fun request2(): String {
delay(2 * 1000)
Log.e(TAG, "request2 completed")
return "result from request2"
}
}
1.将 CoroutineScene2.kt 代码进行反编译之后
public final class CoroutineScene2 {
private static final String TAG = "CoroutineScene2";
@NotNull
public static final CoroutineScene2 INSTANCE;
//1.经过反编译后生成的java代码自动帮我们传入了一个 Continuation 对象(可看做 callback 回调)
@Nullable
public final Object request1(@NotNull Continuation var1) {
Object $continuation;
label20: {
//14.进行条件判断，并将 lable 值改为 1，使得其进入到 case:1分支
if (var1 instanceof <undefinedtype>) {
$continuation = (<undefinedtype>)var1;
if ((((<undefinedtype>)$continuation).label & Integer.MIN_VALUE) != 0) {
((<undefinedtype>)$continuation).label -= Integer.MIN_VALUE;
break label20;
}
}
//2.将传递的接口进行包装，当协程恢复的时候会回调 invokeSuspend 方法。然后会再次进入到 request1 方法。第二次进入的时候会直接执行上方的
//if分支，然后break。
$continuation = new ContinuationImpl(var1) {
// $FF: synthetic field
Object result;
int label;
//13.此时 request2中的结果会回调到这里。该方法返回的结果为 “request2 completed”，并且改变 lable 数值，使得再次调用 request1方法后会进入 if 分支。
@Nullable
public final Object invokeSuspend(@NotNull Object $result) {
this.result = $result;
this.label |= Integer.MIN_VALUE;
return CoroutineScene2.this.request1(this);
}
};
}
//3.首次进入 request1() 方法，lable 为0，会走 case:0 分支
Object $result = ((<undefinedtype>)$continuation).result;
Object var5 = IntrinsicsKt.getCOROUTINE_SUSPENDED();
Object var10000;
switch(((<undefinedtype>)$continuation).label) {
case 0:
ResultKt.throwOnFailure($result);
((<undefinedtype>)$continuation).label = 1;
//4.然后去调用 request2 方法，并将包装的Continuation接口传递进去，用于后面恢复的回调
//8.由步骤7知道当前返回结果为 COROUTINE_SUSPENDED，表明该函数是一个挂起函数，然后 return COROUTINE_SUSPENDED,至此协程方法挂起执行完成
var10000 = this.request2((Continuation)$continuation);
if (var10000 == var5) {
return var5;
}
break;
//15.未return
case 1:
ResultKt.throwOnFailure($result);
var10000 = $result;
break;
default:
throw new IllegalStateException("call to 'resume' before 'invoke' with coroutine");
}
//16.至此整个 suspend 函数执行完成，返回最终结果。
String request2 = (String)var10000;
return "result from request1 " + request2;
}
@Nullable
public final Object request2(@NotNull Continuation var1) {
Object $continuation;
label20: {
if (var1 instanceof <undefinedtype>) {
$continuation = (<undefinedtype>)var1;
if ((((<undefinedtype>)$continuation).label & Integer.MIN_VALUE) != 0) {
((<undefinedtype>)$continuation).label -= Integer.MIN_VALUE;
break label20;
}
}
//5.第一次进入 request2()方法会也对Continuation进行包装，用于request2()挂起方法回复后那这个Continuation进行回调
$continuation = new ContinuationImpl(var1) {
// $FF: synthetic field
Object result;
int label;
@Nullable
public final Object invokeSuspend(@NotNull Object $result) {
//10.此时的label 为步骤7中赋值的1，并且将 labe 与 Integer.MIN_VALUE做或运算，然后再次调用 request2 方法。进入到 label20 的 if分支，将 label 转为 1，进入到下面的 case:1 分支。
this.result = $result;
this.label |= Integer.MIN_VALUE;
return CoroutineScene2.this.request2(this);
}
};
}
//6.第一次进入，此时 label 为0，走 case:0分支。
Object $result = ((<undefinedtype>)$continuation).result;
Object var4 = IntrinsicsKt.getCOROUTINE_SUSPENDED();
switch(((<undefinedtype>)$continuation).label) {
case 0:
ResultKt.throwOnFailure($result);
//将 lable 进行重置，避免再次进入该分支
((<undefinedtype>)$continuation).label = 1;
//7.调用 DelayKt.delay 方法，该方法是一个挂机函数，会返回 COROUTINE_SUSPENDED，此时 if 条件成立，return COROUTINE_SUSPENDED
if (DelayKt.delay(2000L, (Continuation)$continuation) == var4) { //9.delay 方法延迟 2s 后恢复，通过Continuation接口回调接口中的 request2() 中 ContinuationImpl#invokeSuspend 方法
return var4;
}
break;
//11.做异常判断，注意此case分支没有做 return 操作
case 1:
ResultKt.throwOnFailure($result);
break;
default:
throw new IllegalStateException("call to 'resume' before 'invoke' with coroutine");
}
//12.此时 request2方法中的 delay(2 * 1000) 方法之后的代码才真正得到执行。request2方法被恢复后返回 result from request2 结果。因为 request2 中的 Continuation中持有 request1 传入的Continuation接口，在恢复的时候会层级递归恢复，会回调 ContinuationImpl#invokeSuspend方法
Log.e(TAG, "request2 completed");
return "result from request2";
}
private CoroutineScene2() {
}
static {
CoroutineScene2 var0 = new CoroutineScene2();
INSTANCE = var0;
TAG = "CoroutineScene2";
}
}
? 整个函数的挂起和恢复本质就是前面提到的：return + callback 来将我们的异步方法转成同步方法，从 kotlin 代码是看不出来的。需要反编译之后分析。通过方法的递归调用做到挂起和恢复。
2.用Java协程的挂起与恢复
? 上述 CoroutineScene2.kt 中的方法被 suspend 方法修饰后就变成支持挂起的函数，是否被挂起还需要看其内部是否有挂起函数。下面用 Java 代码来复现一下CoroutineScene2.kt 反编译后的 Java代码的执行流程。这样可以直接通过 Debug 进行调试。
public class CoroutineScene2_decompiled {
private static final String TAG = "CoroutineScene2";
public static final Object request1(Continuation preCallback) {
ContinuationImpl request1Callback;
//1.第一次进入 request1 方法时候进入if 分支。对 preCallback 进行封装。
if (!(preCallback instanceof ContinuationImpl) || (((ContinuationImpl) preCallback).label & Integer.MIN_VALUE) == 0) {
request1Callback = new ContinuationImpl(preCallback) {
@Override
Object invokeSuspend(@NotNull Object resumeResult) {
//13.步骤类似，此时修改 label 标签，再次调用 request1 方法
this.result = resumeResult;
this.label |= Integer.MIN_VALUE;
Log.e(TAG, "request1 has resumed " +Thread.currentThread().getName() );
return request1(this);
}
};
} else {
request1Callback = (ContinuationImpl) preCallback;
}
//2.初始状态下 ContinuationImpl.label 值为 0，进入 case:0 分支
switch (request1Callback.label) {
case 0: {
//3.执行 request2 方法，并将封装的 Continuation 对象传过去。
Object request2 = request2(request1Callback);
//6.request2 但会的结果为 COROUTINE_SUSPENDED，此时会进入 if 分支，request1 方法返回 COROUTINE_SUSPENDED，函数被挂起。
if (request2 == IntrinsicsKt.getCOROUTINE_SUSPENDED()) {
Log.e(TAG, "request1 has suspended " + Thread.currentThread().getName());
return IntrinsicsKt.getCOROUTINE_SUSPENDED();
}
}
}
//14.此时 request1 执行到这里返回结果。
Log.e(TAG, "request1 completed");
return "result from request1" + request1Callback.result;
}
public static final Object request2(Continuation preCallback) {
//4.第一次进入 request2 方法，因为label = 0，会进入 if 分支，创建 ContinuationImpl 对象。
ContinuationImpl request2Callback;
if (!(preCallback instanceof ContinuationImpl) || (((ContinuationImpl) preCallback).label & Integer.MIN_VALUE) == 0) {
request2Callback = new ContinuationImpl(preCallback) {
@Override
Object invokeSuspend(@NotNull Object resumeResult) {
//9.此时将 label 标识改为 Integer.MIN_VALUE，防止再次调用 request2 方法 再次进入到 if 分支，而应该走 else 分支。
this.result = resumeResult;
this.label |= Integer.MIN_VALUE;
Log.e(TAG, "request2 has resumed " + Thread.currentThread().getName());
return request2(this);
}
};
} else {
//10.第二次执行 request2 方法会走这里
request2Callback = (ContinuationImpl) preCallback;
}
switch (request2Callback.label) {
case 0: {
//5. 调用 request2 中的 delay(2 * 1000) 方法，该函数是一个 suspend 修饰的挂起函数，并将 request2Callback 传递进去，此方法会马上返回一个 COROUTINE_SUSPENDED 标识，进入到 if 分支中返回
Object delay = DelayKt.delay(2000, request2Callback);//7.delay函数在延迟 2s 后会通过传入的 request2Callback 执行 Continuation#resumeWith 方法。
if (delay == IntrinsicsKt.getCOROUTINE_SUSPENDED()) {
Log.e(TAG, "request2 has suspended" + Thread.currentThread().getName());
return IntrinsicsKt.getCOROUTINE_SUSPENDED();
}
}
}
//11.此时 request 方法被恢复，返回结果。结果返回步骤 8 的 invokeSuspend 方法。
Log.e(TAG, "request2 completed");
return "result from request2";
}
static abstract class ContinuationImpl<T> implements Continuation<T> {
private Continuation preCallback;
int label;
Object result;
public ContinuationImpl(Continuation preCallback) {
this.preCallback = preCallback;
}
@NotNull
@Override
public CoroutineContext getContext() {
return preCallback.getContext();
}
@Override
public void resumeWith(@NotNull Object resumeResult) {
//8.此时会执行 request2中ContinuationImpl.invokeSuspend 方法
Object suspend = invokeSuspend(resumeResult);
//12.得到 request2 的返回结果 "result from request2" 后，执行 preCallback.resumeWith(suspend)方法，又会走到 request1 中的ContinuationImpl.invokeSuspend方法。
//15.得到 request1 中的方法返回的结果为 "result from request1 result from request2",最终调用我们传入的 Continuation 对象的 resumeWith 方法。
if (suspend == IntrinsicsKt.getCOROUTINE_SUSPENDED()) {
return;
}
preCallback.resumeWith(suspend);
}
abstract Object invokeSuspend(@NotNull Object resumeResult);
}
}
在Java方法中调用上述的 reques1() 方法，传入 Continuation 接口对象：
日志结果：
? 从 Java 代码中可以明显感觉到，所谓的挂起与恢复本质上就是一个 return 和 callback 的方式。可以按照上图代码标注的步骤打断点进行调试。能更加清晰的理解这协程挂起与恢复的流程。