概述
接下来我会围绕一个问题探讨一下 withTimeout 的实现，这个问题就是：为什么 Thread.sleep 在 withTimeout 中无效？
1. 协程的取消协作机制
在 Kotlin 的官方文档中，说到了协程的取消是协作完成的，这也是为什么协程叫协程，协程的代码必须进行协作才能被取消，而 withTimeout 也是通过这套取消协作机制完成的，比如下面的例子中，在打印第四个数字的时候，就在执行超时后抛出了异常。
withTimeout(1300L) {
repeat(1000) { i ->
println("I'm sleeping $i ...")
delay(500L)
}
}
上面这段代码的执行结果如下：
I'm sleeping 0 ...
I'm sleeping 1 ...
I'm sleeping 2 ...
Exception in thread "main" kotlinx.coroutines.TimeoutCancellationException: Timed out waiting for 1300 ms
withTimeout 抛出的 TimeoutCancellationException 是 CancellationException 的子类，由于取消只是一个异常，所有的资源都可以按照通常的方式关闭。如果需要在任何类型的超时发生时执行某些额外的操作，则可以将代码包装在 try{...} catch(）{...} 块中。或者可以使用 withTimeoutOrNull 函数，该函数类似于 withTimeout，但在超时时返回 null 而不是抛出异常：
val result = withTimeoutOrNull(1300L) {
repeat(1000) { i ->
println("I'm sleeping $i ...")
delay(500L)
}
"Done" // will get cancelled before it produces this result
}
println("Result is $result")
测试代码 1 ：withTimeout() 与 delay()
class ExampleUnitTest {
@Test
fun testWithTimeoutAndDelay() {
val latch = CountDownLatch(1)
GlobalScope.launch {
kotlin.runCatching {
// 20 秒内没有执行完的话就抛出超时取消异常
withTimeout(20 * 1000) {
// 延迟 30 秒
delay(30 * 1000)
println("after delay")
latch.countDown()
}
}.onFailure {
println("failed: $it")
}
}
latch.await()
}
}
上面这段代码对应的 Java 代码如下：
public final class ExampleUnitTest {
@Test
public final void testWithTimeoutAndDelay() {
final Ref.ObjectRef latch = new Ref.ObjectRef();
latch.element = new CountDownLatch(1);
// launch 代码块
BuildersKt.launch$default((CoroutineScope)GlobalScope.INSTANCE, (CoroutineContext)Dispatchers.getUnconfined(), (CoroutineStart)null, (Function2)(new Function2((Continuation)null) {
int label;
@Nullable
public final Object invokeSuspend(@NotNull Object $result) {
Throwable var3;
Object var12;
Throwable var10000;
label42: {
Result.Companion var13;
label41: {
Object var8 = IntrinsicsKt.getCOROUTINE_SUSPENDED();
boolean var2;
boolean var10001;
switch (this.label) {
case 0:
// 如果 result 是 Result.Failure ，则抛出 Failure 的异常值
ResultKt.throwOnFailure($result);
Object var14;
try {
var13 = Result.Companion;
var2 = false;
// 创建并执行 withTimeout 代码块
Function2 var15 = (Function2)(new ExampleUnitTest$testWithTimeoutAndDelay$1$invokeSuspend$$inlined$runCatching$lambda$1((Continuation)null, this));
this.label = 1;
var14 = TimeoutKt.withTimeout(20000L, var15, this);
} catch (Throwable var10) {
var10000 = var10;
var10001 = false;
break label41;
}
if (var14 == var8) {
return var8;
}
break;
case 1:
var2 = false;
try {
ResultKt.throwOnFailure($result);
break;
} catch (Throwable var11) {
var10000 = var11;
var10001 = false;
break label41;
}
default:
throw new IllegalStateException("call to 'resume' before 'invoke' with coroutine");
}
try {
var12 = Result.constructor-impl(Unit.INSTANCE);
break label42;
} catch (Throwable var9) {
var10000 = var9;
var10001 = false;
}
}
var3 = var10000;
var13 = Result.Companion;
var12 = Result.constructor-impl(ResultKt.createFailure(var3));
}
var10000 = Result.exceptionOrNull-impl(var12);
if (var10000 != null) {
var3 = var10000;
int var6 = false;
String var7 = "failed: " + var3;
System.out.println(var7);
}
return Unit.INSTANCE;
}
@NotNull
public final Continuation create(@Nullable Object value, @NotNull Continuation completion) {
Intrinsics.checkNotNullParameter(completion, "completion");
Function2 var3 = new <anonymous constructor>(completion);
return var3;
}
public final Object invoke(Object var1, Object var2) {
return ((<undefinedtype>)this.create(var1, (Continuation)var2)).invokeSuspend(Unit.INSTANCE);
}
}), 2, (Object)null);
((CountDownLatch)latch.element).await();
}
}
// ExampleUnitTest$testWithTimeoutAndDelay$1$invokeSuspend$$inlined$runCatching$lambda$1.java
// ...
@DebugMetadata(
f = "ExampleUnitTest.kt",
l = {24},
i = {},
s = {},
n = {},
m = "invokeSuspend",
c = "com.example.myapplication.ExampleUnitTest$testWithTimeoutAndDelay$1$1$1"
)
// withTimeout 代码块
final class ExampleUnitTest$testWithTimeoutAndDelay$1$invokeSuspend$$inlined$runCatching$lambda$1 extends SuspendLambda implements Function2 {
// 状态机
int label;
// $FF: synthetic field
final <undefinedtype> this$0;
ExampleUnitTest$testWithTimeoutAndDelay$1$invokeSuspend$$inlined$runCatching$lambda$1(Continuation var1, Object var2) {
super(2, var1);
this.this$0 = var2;
}
@Nullable
public final Object invokeSuspend(@NotNull Object $result) {
Object var2 = IntrinsicsKt.getCOROUTINE_SUSPENDED();
switch (this.label) {
case 0:
// 如果 result 是 Result.Failure ，则抛出 Failure 的异常值
ResultKt.throwOnFailure($result);
// 迁移到下一个状态，下一次执行走 case 1
this.label = 1;
if (DelayKt.delay(30000L, this) == var2) {
return var2;
}
break;
case 1:
// 如果 result 是 Result.Failure ，则抛出 Failure 的异常值
ResultKt.throwOnFailure($result);
break;
default:
throw new IllegalStateException("call to 'resume' before 'invoke' with coroutine");
}
String var4 = "after delay";
System.out.println(var4);
((CountDownLatch)this.this$0.$latch.element).countDown();
return Unit.INSTANCE;
}
@NotNull
public final Continuation create(@Nullable Object value, @NotNull Continuation completion) {
Intrinsics.checkNotNullParameter(completion, "completion");
ExampleUnitTest$testWithTimeoutAndDelay$1$invokeSuspend$$inlined$runCatching$lambda$1 var3 = new ExampleUnitTest$testWithTimeoutAndDelay$1$invokeSuspend$$inlined$runCatching$lambda$1(completion, this.this$0);
return var3;
}
public final Object invoke(Object var1, Object var2) {
return ((ExampleUnitTest$testWithTimeoutAndDelay$1$invokeSuspend$$inlined$runCatching$lambda$1)this.create(var1, (Continuation)var2)).invokeSuspend(Unit.INSTANCE);
}
}
在上面这段代码中，具体会抛出超时取消异常的地方就在 ...$inlined$runCatching$lambda$1 匿名内部类的 invokeSuspend 方法的第二个 case 语句，可以看到，每一个续体（SuspendLambda）的状态机的第一行都是 ResultKt.throwOnFailure($result) ，这行代码会检查接收到的上一个续体的执行结果是否为 Failure ，如果是的话，则抛出异常，下面再来看下 Thread.sleep() 的 Java 代码的区别。
测试代码 2 ：withTimeout() 与 Thread.sleep()
class ExampleUnitTe {
@Test
fun testWithTimeoutAndThreadSleep() {
val latch = CountDownLatch(1)
GlobalScope.launch(Dispatchers.Unconfined) {
kotlin.runCatching {
// 两秒钟后没有执行完成则抛出异常
withTimeout(2 * 1000) {
// 让当前线程进入阻塞状态
Thread.sleep(4 * 1000)
println("after sleep")
}
}.onFailure {
println("failed: $it")
latch.countDown()
}
}
latch.await()
println("end")
}
}
上面这段代码对应的 Java 代码如下。
public final class ExampleUnitTest2 {
@Test
public final void testWithTimeoutAndThreadSleep() {
final CountDownLatch latch = new CountDownLatch(1);
BuildersKt.launch$default((CoroutineScope)GlobalScope.INSTANCE, (CoroutineContext)Dispatchers.getUnconfined(), (CoroutineStart)null, (Function2)(new Function2((Continuation)null) {
int label;
@Nullable
public final Object invokeSuspend(@NotNull Object $result) {
Throwable var3;
Object var12;
Throwable var10000;
label42: {
Result.Companion var13;
label41: {
Object var8 = IntrinsicsKt.getCOROUTINE_SUSPENDED();
boolean var2;
boolean var10001;
switch (this.label) {
case 0:
ResultKt.throwOnFailure($result);
Object var14;
try {
var13 = Result.Companion;
var2 = false;
// 创建并执行 withTimeout 代码块
Function2 var15 = (Function2)(new ExampleUnitTest2$testWithTimeoutAndThreadSleep$1$1$1((Continuation)null));
this.label = 1;
var14 = TimeoutKt.withTimeout(2000L, var15, this);
} catch (Throwable var10) {
var10000 = var10;
var10001 = false;
break label41;
}
if (var14 == var8) {
return var8;
}
break;
case 1:
var2 = false;
try {
ResultKt.throwOnFailure($result);
break;
} catch (Throwable var11) {
var10000 = var11;
var10001 = false;
break label41;
}
default:
throw new IllegalStateException("call to 'resume' before 'invoke' with coroutine");
}
try {
var12 = Result.constructor-impl(Unit.INSTANCE);
break label42;
} catch (Throwable var9) {
var10000 = var9;
var10001 = false;
}
}
var3 = var10000;
var13 = Result.Companion;
var12 = Result.constructor-impl(ResultKt.createFailure(var3));
}
var10000 = Result.exceptionOrNull-impl(var12);
if (var10000 != null) {
var3 = var10000;
int var6 = false;
String var7 = "failed: " + var3;
System.out.println(var7);
latch.countDown();
}
return Unit.INSTANCE;
}
@NotNull
public final Continuation create(@Nullable Object value, @NotNull Continuation completion) {
Intrinsics.checkNotNullParameter(completion, "completion");
Function2 var3 = new <anonymous constructor>(completion);
return var3;
}
public final Object invoke(Object var1, Object var2) {
return ((<undefinedtype>)this.create(var1, (Continuation)var2)).invokeSuspend(Unit.INSTANCE);
}
}), 2, (Object)null);
latch.await();
String var2 = "end";
System.out.println(var2);
}
}
// ExampleUnitTest2$testWithTimeoutAndThreadSleep$1$1$1.java
// ...
@Metadata(
mv = {1, 8, 0},
k = 3,
d1 = {"\u0000\u000e\n\u0000\n\u0002\u0010\u0002\n\u0002\u0018\u0002\n\u0002\b\u0002\u0010\u0000\u001a\u00020\u0001*\u00020\u0002H\u008a@¢\u0006\u0004\b\u0003\u0010\u0004"},
d2 = {"<anonymous>", "", "Lkotlinx/coroutines/CoroutineScope;", "invoke", "(Ljava/lang/Object;Ljava/lang/Object;)Ljava/lang/Object;"}
)
final class ExampleUnitTest2$testWithTimeoutAndThreadSleep$1$1$1 extends SuspendLambda implements Function2 {
int label;
@Nullable
public final Object invokeSuspend(@NotNull Object var1) {
Object var3 = IntrinsicsKt.getCOROUTINE_SUSPENDED();
switch (this.label) {
case 0:
ResultKt.throwOnFailure(var1);
Thread.sleep(4000L);
String var2 = "after sleep";
System.out.println(var2);
return Unit.INSTANCE;
default:
throw new IllegalStateException("call to 'resume' before 'invoke' with coroutine");
}
}
ExampleUnitTest2$testWithTimeoutAndThreadSleep$1$1$1(Continuation var1) {
super(2, var1);
}
@NotNull
public final Continuation create(@Nullable Object value, @NotNull Continuation completion) {
Intrinsics.checkNotNullParameter(completion, "completion");
ExampleUnitTest2$testWithTimeoutAndThreadSleep$1$1$1 var3 = new ExampleUnitTest2$testWithTimeoutAndThreadSleep$1$1$1(completion);
return var3;
}
public final Object invoke(Object var1, Object var2) {
return ((ExampleUnitTest2$testWithTimeoutAndThreadSleep$1$1$1)this.create(var1, (Continuation)var2)).invokeSuspend(Unit.INSTANCE);
}
}
上面这段代码，因为没有调用 delay ，匿名内部类ExampleUnitTest2$testWithTimeoutAndThreadSleep$1$1$1 的状态机只有一种状态，而第一次执行第一个 case 语句的时候，还没有超时，后面也没有下一个状态继续检查续体（SuspendLambda）接收到的上一个续体（Continuation）的执行结果了。
代码对比：
反编译后的代码对比.png
任务提交与异常抛出流程
withTimeout+与+delay+的任务提交与执行.png
超时取消异常抛出的流程大致如上，launch 代码块会创建独立协程 StandaloneCoroutine ，再由独立协程执行 withTimeout 代码块。withTimeout 会创建超时取消协程 TimeoutCoroutine，并通过事件轮询器 DefaultExecutor 创建并提交延时执行任务 DelayedRunanbleTask ，最后通过 delay 方法创建可取消续体 CancellableContinuationImpl 和延时恢复任务 DelayedResumeTask，并把延时恢复任务提交到事件轮询器。在 CancellableCoroutineImpl 初始化的时候，会把自己设为父任务（超时协程）的工作节点，这个点很关键，因为只有设为工作节点后，父任务被取消或者完成的时候，它才能收到通知，也就是 withTimeout 是给其他的续体兜底的，如果没有其他的协程或续体在执行的话，那它的超时就不会起作用。
因为协程的这套协作机制，我们在使用协程的过程中，就尽量不要混着线程池、 RxJava 或其他异步框架和或响应式框架来使用协程，而是尽量使用协程 + Flow ，这样才能让协程之间更好地协作。
在事件轮询器中，有一个专门用来轮询任务的线程，并且维护了一个延时任务队列和一个普通任务队列，当到了延迟的时间后，没有任务的时候，负责轮询任务的线程就会被阻塞，直到下一个延时任务的时间到了后，就会被唤醒，这时执行 延时任务，由于在前面的第一个单测例子中，delay 的时间比 withTimeout 要长，在还没有执行完 delay 的时候，withTimeout 创建的延时运行任务会先执行，它执行时会调用超时取消协程的 run() 方法，超时取消协程就会通过 cancelCoroutine() 方法把 TimeoutCancellationException 作为结果通知给工作节点，也就是 delay 创建的可取消续体 ，然后可取消续体就会恢复 withTimeout 对应的 SuspendLambda 的执行，也就是走到了反编译后的匿名内部类的第二个状态，检查发现有异常后，就会抛出异常。
而前面提到的第二个调用了 Thread.sleep 的单测中，由于没有执行 delay ，也就没有创建可取消续体，超时取消协程运行时，并没有其他的工作节点需要通知，所以 withTimeout 代码块不会再次执行，也没有第二个状态，所以在 withTimeout 中执行 Thread.sleep 就不会抛出超时取消异常。
withTimeout+的恢复.png
如上图所示，在只调用了 Thread.sleep 时，只有 launch 和 withTimeout 两个续体（一个协程同时是一个实现了 Continuation 接口的类），有 delay 时，如果没有超时，则由 delay 来通知事件轮询器，让事件轮询器恢复 withTimeout 代码块的执行，事件轮询器可以看成是协程框架自定义的 Looper 。如果超时了还没有执行完 withTimeout 代码块的代码，则由事件轮询器来通知 DelayedRunnableTask ，让它通过超时协程来恢复 withTimeout 的第二个状态的执行，同时执行结果为 Failure 。
（续体是一个协程的恢复点，当协程代码块因为调用了挂起函数被分为多个状态后，就需要续体来恢复一个个状态的执行）。
超时协程的工作节点
超时取消协程的工作节点.png
更具体一点来说，前面提到的两个单测例子创建的工作节点主要是 DisposeOnCompletion 和 CancellableContinuatonImpl 。
第二个单测中，只调用了 Thread.sleep ，没有调用 delay ，那超时协程就只有一个工作节点 DisposeOnCompletion ，这个工作结点是 withTimeout 方法中创建的。DisposeOnCompletion 持有了延时运行任务 DelayedRunnableTask 作为句柄，当它收到完成协程执行完成回调的时候，就会把延时运行任务的状态改为 REMOVED_TASK ，这样延时任务就无法再被调度。
如果调用了 delay ，delay 会通过 suspendCancellableCoroutine 方法创建可取消续体，可取消续体在初始化可取消性的时候，会创建子续体 ChildContinuation ，并把子续体作为超时取消协程的工作节点，这样超时协程在超时取消的时候，就会通知给可取消续体 ，把异常传给 withTimeout 代码块对应的 SuspendLambda 续体，从而抛出异常。
下面来看下部分源码的解析，由于有些源码在我上一篇文章中有解析过，所以这里就不再重复说明了。
2. withTimeout
独立协程与超时协程的初始化时序图.png
由于 launch 方法的实现在我的上一篇文章（探索 Kotlin 协程原理）中有讲过，所以这里就不细说了，从第一个单元测试反编译后的 Java 代码来看，执行 launch 代码会实例化一个 SuspendLambda 对象，SuspendLambda 会被封装为 DispatchedContinuation ，DispatchedContinuation 是 DispatchedTask 的子类，会被提交到协程分发器中，从而执行挂起 Lambda 表达式（SuspendLambda）中的代码。
在 launch 创建了独立协程 StandaloneCoroutine 后，就会通过 withTimeout 方法创建超时协程 TimeoutCorotuine 。
2.1 withTimeout
withTimeout 方法用于在指定的超时时间内，在一个协程中运行给定的挂起代码，并在超时时抛出 TimeoutCancellationException 超时取消异常，执行在 withTimeout 中的代码在超时时将被取消。如果不想在超时的时候抛出异常，可以使用 withTimeoutOrNull 方法。
/**
* @param timeMillis 超时时间，单位为毫秒。
*/
public suspend fun <T> withTimeout(timeMillis: Long, block: suspend CoroutineScope.() -> T):
T {
// ...
// 如果时间小于等于 0 ，则立刻跑出异常
if (timeMillis <= 0L) throw TimeoutCancellationException("Timed out immediately")
// 通过挂起协程运行代码块，如果超时了就抛出 TimeoutCancellationException 异常。
return suspendCoroutineUninterceptedOrReturn { uCont ->
setupTimeout(TimeoutCoroutine(timeMillis, uCont), block)
}
}
TimeoutCoroutine
private class TimeoutCoroutine<U, in T: U>(
@JvmField val time: Long,
uCont: Continuation<U> // 未拦截的续体
) : ScopeCoroutine<T>(uCont.context, uCont), Runnable {
override fun run() {
// 取消协程
cancelCoroutine(TimeoutCancellationException(time, this))
}
// ...
}
TimeoutCoroutine 是 ScopeCoroutine 的子类，ScopeCoroutine 是通过 coroutineScope 协程构建器创建的协程实例。
作用域协程 ScopeCoroutine
续体相关类图.png
ScopeCoroutine 除了 TimeoutCoroutine 还有其他的子类，比如 withContext 方法中会判断协程上下文（协程分发器）是否发生了变化，如果没有变化的话，则使用 UndispatchedCoroutine 直接在当前线程执行协程代码，如果发生了变化，则使用 DispatchedCoroutine 把协程代码块切换到另一个分发器执行。
另外通过 Flow.shareIn 扩展函数，也可以创建 FlowCoroutine，用 supervisorScope() 函数则可以创建 SupervisorCoroutine ，SupervisorCoroutine 会把协程上下文中原来的 Job 替换为 SupervisorJob ，从而让代码块内的某个子协程在失败时不影响其他子协程。
TimeoutCoroutine 还实现了 Runnable 接口，它的 run 方法是通过协程的事件轮询机制（EventLoop）来触发的，在 run 方法中，创建了一个 TimeoutCancellationException ，并传给了 cancelCoroutine() 方法，这个异常会被封装为表示协程执行失败的 Result.Failure ，前面有讲到，只要上一个续体的执行结果是 Failure ，SuspendLambda 的第二个 case 语句就会检查到异常并抛出。
setupTimeout
private fun <U, T: U> setupTimeout(
coroutine: TimeoutCoroutine<U, T>,
block: suspend CoroutineScope.() -> T
): Any? {
// 协程的续体
val cont = coroutine.uCont
val context = cont.context
// 创建延时执行任务，并设置超时取消协程的完成回调
coroutine.disposeOnCompletion(context.delay.invokeOnTimeout(coroutine.time, coroutine, coroutine.context))
// 在当前线程执行代码块
return coroutine.startUndispatchedOrReturnIgnoreTimeout(coroutine, block)
}
在 setupTimeout 方法中，会调用协程上下文的 delay 对象的 invokeOnTimout 方法，delay 对象的默认实现是 DefaultExecutor， 实现了 Delay 接口的有下面几个类。
image.png
-
BlockingEventLoop ：用于在
runBlocking中处理任务，包括延时任务 -
DefaultExecutor ：用于在 IO 线程和默认的子线程中执行任务，包括延时任务
-
Dispatcher in TestCoroutineContext ：用于在负责执行测试的线程中处理任务，包括延时任务
-
EventLoopImplBase ：
BlockingEventLoop和DefaultExecutor的父类 -
ExecutorCoroutineDispatcherImpl ：
用于在自定义的协程分发器（通过
ExecutorService.asCoroutineDispatcher创建）中处理任务，包括延时任务 -
HandlerContext：
就是平常使用的
Dispatchers.Main的实现类，是HandlerDispatcher的子类，用于在主线程中执行任务，包括延时任务，通过主线程 Looper 实现，也可以通过Handler.asCoroutineDispatcher把自定义的 Handler 转换为HandlerContext
Delay#invokeOnTimeout
/**
* [CoroutineDispatcher]实现以原生支持任务的定时执行功能, 实现此接口会影响[delay]和[withTimeout]函数的操作。
*
* @suppress **这是一个内部 API，不应从一般代码中使用。**
*/
@InternalCoroutinesApi
public interface Delay {
/**
* 不阻塞线程并在指定的时间后继续暂停协程。
*
* 这个挂起函数可以被取消。如果当前协程的[Job]在该挂起函数等待期间被取消或完成，
* 此函数将立即使用[CancellationException]继续执行。有**即时取消保证**，如果该作业在该函数挂起时被取消，则它将无法成功恢复。
* 有关详细信息，请参见[suspendCancellableCoroutine]文档。
*/
public suspend fun delay(time: Long) {
if (time <= 0) return // 不需要延迟
return suspendCancellableCoroutine { scheduleResumeAfterDelay(time, it) }
}
// ...
/**
* 在指定的延迟[timeMillis]后调度[Runnable]的[block]调用。如果不再需要这个调用请求，那么产生的[DisposableHandle]可以被用于[dispose][DisposableHandle.dispose].
*
* 这个实现使用一个内置的单线程计划的 executor 服务。
*/
public fun invokeOnTimeout(timeMillis: Long, block: Runnable, context: CoroutineContext): DisposableHandle =
DefaultDelay.invokeOnTimeout(timeMillis, block, context)
}
invokeOnTimeout() 方法会调用 DefaultDelay 的 invokeOnTimeout 的方法。
DefaultExecutor#invokeOnTimeout
internal actual val DefaultDelay: Delay = DefaultExecutor
@Suppress("PLATFORM_CLASS_MAPPED_TO_KOTLIN")
internal actual object DefaultExecutor : EventLoopImplBase(), Runnable {
override fun invokeOnTimeout(timeMillis: Long, block: Runnable, context: CoroutineContext): DisposableHandle =
scheduleInvokeOnTimeout(timeMillis, block)
}
DefaultExecutor 的 invokeOnTimeout 方法会调用父类 EventLoopImplBase 的 scheduleInvokeOnTimeout 方法，通过该方法把任务加入到延迟任务队列。
scheduleInvokeOnTimeout
internal abstract class EventLoopImplBase: EventLoopImplPlatform(), Delay {
// 调度即将到来的任务
protected fun scheduleInvokeOnTimeout(
timeMillis: Long,
block: Runnable
): DisposableHandle {
// 将毫秒数转换为纳秒数
val timeNanos = delayToNanos(timeMillis)
// 如果时间不超过最大延时，则创建一个 DelayedRunnableTask 任务，并调度该任务
return if (timeNanos < MAX_DELAY_NS) {
val now = nanoTime()
// 创建延时可运行任务
DelayedRunnableTask(now + timeNanos, block).also { task ->
schedule(now, task) // 调度传入的任务
}
} else {
// 如果超过了最大延时，则返回 NonDisposableHandle
NonDisposableHandle
}
}
// 把延时任务添加到延时任务队列
public fun schedule(now: Long, delayedTask: DelayedTask) {
when (scheduleImpl(now, delayedTask)) {
SCHEDULE_OK -> if (shouldUnpark(delayedTask)) unpark()
SCHEDULE_COMPLETED -> reschedule(now, delayedTask)
SCHEDULE_DISPOSED -> {} // do nothing -- task was already disposed
else -> error("unexpected result")
}
}
private fun scheduleImpl(now: Long, delayedTask: DelayedTask): Int {
if (isCompleted) return SCHEDULE_COMPLETED
// 如果延时队列为空，则创建一个新的延时任务队列
val delayedQueue = _delayed.value ?: run {
_delayed.compareAndSet(null, DelayedTaskQueue(now))
_delayed.value!!
}
// 把任务添加到队列
return delayedTask.scheduleTask(now, delayedQueue, this)
}
// 延时运行任务
private class DelayedRunnableTask(
nanoTime: Long, // 任务的执行时间
private val block: Runnable // 代码块（任务的具体工作）
) : DelayedTask(nanoTime) {
override fun run() { block.run() } // 运行任务中的 Runnable
}
}
scheduleInvokeOnTimeout 方法会创建一个 DelayedRunnableTask ，并把该 Task 通过 schedule 方法加入到延时任务队列。DelayedRunnableTask 是 DelayedTask 的子类，DelayedTask 的另一个子类是 DelayedReumeTask ，DelayedResumeTask 就是调用 delay() 方法时需要用到的任务类型。
schedule 的具体实现是 scheduleImpl ，scheduleImpl 方法中会判断 _delayed 延时任务队列是否为空，为空的话则创建一个延时任务队列 DelayedTaskQueue ，再把延时任务（超时回调）通过 DelayedTask 的 scheduleTask 方法放到延时任务队列中，scheduleImpl 返回调度成功后，就会唤醒事件轮询。
DelayedTask
// 延时任务
internal abstract class DelayedTask(
/** 任务的执行时间 */
@JvmField var nanoTime: Long
) : Runnable, Comparable<DelayedTask>, DisposableHandle, ThreadSafeHeapNode {
/**
* _heap 变量可以是三种类型：null、ThreadSafeHeap 或 DISPOSED_TASK。
*/
private var _heap: Any? = null
/**
* 判断是不是要执行任务了，当被调用时，将给定时间 now 与当前对象的 nanoTime 属性进行比较。
* 如果 now >= nanoTime，则返回 true；否则返回 false。
*/
fun timeToExecute(now: Long): Boolean = now - nanoTime >= 0L
/**
* scheduleTask 方法用于将当前 DelayedTask 对象加入到 DelayedTaskQueue 堆中，以便在给定的时间点执行任务。
* now 表示当前时间，delayed 表示 DelayedTaskQueue 堆，eventLoop 表示 EventLoopImplBase 实例。
* 该方法使用 @Synchronized 注解表示线程同步，保证对 _heap 和 delayed 的修改是原子的。
* 如果 _heap 已经被设置为 DISPOSED_TASK，则返回 SCHEDULE_DISPOSED。
*/
@Synchronized
fun scheduleTask(now: Long, delayed: DelayedTaskQueue, eventLoop: EventLoopImplBase): Int {
if (_heap === DISPOSED_TASK) return SCHEDULE_DISPOSED // don't add -- was already disposed
delayed.addLastIf(this) { firstTask ->
// ...
}
return SCHEDULE_OK
}
/**
* dispose 方法用于移除当前 DelayedTask 对象从 _heap 中，使之不再参与后续堆操作。
* 在 @Synchronized 保护下，首先读取 _heap 的值，如果已经被设置为 DISPOSED_TASK 则直接返回，
* 否则尝试将当前 DelayedTask 从堆中移除。最后将 _heap 的值设置为 DISPOSED_TASK，
* 表示该对象已经被注销，并且不会再添加到任何堆中。
*/
@Synchronized
final override fun dispose() {
val heap = _heap
if (heap === DISPOSED_TASK) return // 已移除
@Suppress("UNCHECKED_CAST")
(heap as? DelayedTaskQueue)?.remove(this) // 从任务堆中移除当前任务
_heap = DISPOSED_TASK // 不再添加到队列中
}
}
DefaultSchedule#run
@Suppress("PLATFORM_CLASS_MAPPED_TO_KOTLIN")
internal actual object DefaultExecutor : EventLoopImplBase(), Runnable {
@Suppress("ObjectPropertyName")
@Volatile
private var _thread: Thread? = null
// 事件轮询线程，如果不存在则创建一个新的轮询线程
override val thread: Thread
get() = _thread ?: createThreadSync()
override fun run() {
ThreadLocalEventLoop.setEventLoop(this)
registerTimeLoopThread()
try {
var shutdownNanos = Long.MAX_VALUE
if (!notifyStartup()) return
while (true) {
Thread.interrupted() // 重置中断标志
// 处理下一个事件
var parkNanos = processNextEvent()
// 没有下一个事件
if (parkNanos == Long.MAX_VALUE) {
// 初始化关闭时间
val now = nanoTime()
// 如果关闭时间是长整型的最大值，则将轮询关闭时间改为一秒后
if (shutdownNanos == Long.MAX_VALUE) shutdownNanos = now + KEEP_ALIVE_NANOS
// 计算剩余需要等待的时间
val tillShutdown = shutdownNanos - now
if (tillShutdown <= 0) return // 时间已过，不再轮询
// 用当前运行的最小值和剩余等待时间中较小的一个来更新等待时间
parkNanos = parkNanos.coerceAtMost(tillShutdown)
} else
shutdownNanos = Long.MAX_VALUE
if (parkNanos > 0) {
// 检查是否请求了关闭线程，是的话退出轮询
if (isShutdownRequested) return
// 不停止运行线程，等待一段时间后唤醒线程
parkNanos(this, parkNanos)
}
}
} finally {
// ...
}
}
}
DefaultExecutor 被唤醒后，就会开始执行 run() 方法，并调用 processNextEvent() 处理下一个事件，在前面提到的第一个单元测试的例子中，下一个事件就是执行延时操作（delay）。
EventLoopImplBase
internal abstract class EventLoopImplBase: EventLoopImplPlatform(), Delay {
// 普通任务队列，可能的值：null | CLOSED_EMPTY | task | Queue<Runnable>
private val _queue = atomic<Any?>(null)
// 延时任务队列
private val _delayed = atomic<DelayedTaskQueue?>(null)
// 处理下一个事件
override fun processNextEvent(): Long {
// 优先处理无限制（不需要调度）的任务
if (processUnconfinedEvent()) return 0
// 再找出延时任务来执行
val delayed = _delayed.value
if (delayed != null && !delayed.isEmpty) {
val now = nanoTime()
while (true) {
delayed.removeFirstIf {
// 判断是不是需要执行该任务
if (it.timeToExecute(now)) {
// 任务的时间到了后，就把任务加入普通任务队列
enqueueImpl(it)
} else
false
} ?: break
}
}
// 取出普通任务并执行
val task = dequeue()
if (task != null) {
task.run()
return 0
}
return nextTime
}
@Suppress("UNCHECKED_CAST")
private fun enqueueImpl(task: Runnable): Boolean {
_queue.loop { queue ->
if (isCompleted) return false // 已停止轮询，不能再添加任务
when (queue) {
// 队列为空，直接把当前任务赋值给 _queue
null -> if (_queue.compareAndSet(null, task)) return true
// 队列不为空，把任务添加到队列
is Queue<*> -> {
when ((queue as Queue<Runnable>).addLast(task)) {
Queue.ADD_SUCCESS -> return true
Queue.ADD_CLOSED -> return false
Queue.ADD_FROZEN -> _queue.compareAndSet(queue, queue.next())
}
}
// _queue 是一个任务的花，则将新任务添加到队尾
else -> when {
queue === CLOSED_EMPTY -> return false
else -> {
val newQueue = Queue<Runnable>(Queue.INITIAL_CAPACITY, singleConsumer = true)
newQueue.addLast(queue as Runnable)
newQueue.addLast(task)
if (_queue.compareAndSet(queue, newQueue)) return true
}
}
}
}
}
// 从普通任务队列中取出任务
@Suppress("UNCHECKED_CAST")
private fun dequeue(): Runnable? {
_queue.loop { queue ->
when (queue) {
null -> return null
is Queue<*> -> {
val result = (queue as Queue<Runnable>).removeFirstOrNull()
if (result !== Queue.REMOVE_FROZEN) return result as Runnable?
_queue.compareAndSet(queue, queue.next())
}
else -> when {
queue === CLOSED_EMPTY -> return null
else -> if (_queue.compareAndSet(queue, null)) return queue as Runnable
}
}
}
}
}
2. delay
delay
/**
* 在不阻塞线程的情况下，延迟协程一段时间并在指定时间后恢复它
*/
public suspend fun delay(timeMillis: Long) {
if (timeMillis <= 0) return // don't delay
return suspendCancellableCoroutine sc@ { cont: CancellableContinuation<Unit> ->
if (timeMillis < Long.MAX_VALUE) {
cont.context.delay.scheduleResumeAfterDelay(timeMillis, cont)
}
}
}
delay 方法会通过 suspendCancellableCoroutine 方法创建一个可取消续体，并把这个续体传给 DefaultExecutor 。
suspendCancellableCoroutine
public suspend inline fun <T> suspendCancellableCoroutine(
crossinline block: (CancellableContinuation<T>) -> Unit
): T =
suspendCoroutineUninterceptedOrReturn { uCont ->
// 创建可取消续体
val cancellable = CancellableContinuationImpl(uCont.intercepted(), resumeMode = MODE_CANCELLABLE)
// 初始化可取消性（设置取消回调）
cancellable.initCancellability()
// 执行代码块
block(cancellable)
// 获取执行结果（把状态改为挂起或直接返回结果）
cancellable.getResult()
}
suspendCancellableCoroutine 中会把上一个续体通过 intercepted() 扩展函数封装为可以分发给分发器执行的 DispatchedContinuation ，再把 DispatchedContinuation 传给续体 CancellableContinuationImpl ，再初始化 CancellableContinuationImpl 的可取消性、执行代码块、获取可取消续体的执行结果。
suspendCancellableCoroutine 方法的典型用法，就是在等待回调结果时挂起协程，并将结果返回给调用方，这样就不用再为回调创建一层嵌套，又或者说是对回调嵌套的封装，如果回调有多次的话，就要用 callbackFlow。
suspend fun awaitCallback(): T = suspendCancellableCoroutine { continuation ->
val callback = object : Callback { // 实现某个回调接口
override fun onCompleted(value: T) {
// 通过回调提供的值恢复协程
continuation.resume(value)
}
override fun onApiError(cause: Throwable) {
// 通过回调提供的异常恢复协程
continuation.resumeWithException(cause)
}
}
// 注册回调
api.register(callback)
// 在取消协程时移除回调
continuation.invokeOnCancellation { api.unregister(callback) }
// 此时 suspendCancellableCoroutine 函数挂起协程，直到回调触发
}
但是要注意的是，suspendCancellableContinuation 的代码块中引用到的续体（continuation）只能恢复一次，如果多次恢复的话，就会抛出异常，所以尽量不要在没有设置异常处理器的线程中恢复续体的执行。
image 1.png
CancellableContinuationImpl#initCancellability
@PublishedApi
internal open class CancellableContinuationImpl<in T>(
final override val delegate: Continuation<T>,
resumeMode: Int
) : DispatchedTask<T>(resumeMode), CancellableContinuation<T>, CoroutineStackFrame {
public override val context: CoroutineContext = delegate.context
/*
* 实现说明
*
* CancellableContinuationImpl 是 Job 的子集，具有以下限制：
* 1）它只能有取消监听器（没有“正在取消”）
* 2）如果被取消，它总是调用取消监听器（没有“立即调用”）
* 3）它最多只能有一个取消监听器
* 4）它的取消监听器无法注销
* 作为结果，它具有更简单的状态机、更轻量级的机制和更少的依赖
*/
/* 决策状态机
+-----------+ trySuspend +-----------+
| UNDECIDED | -------------> | SUSPENDED |
+-----------+ +-----------+
|
| tryResume
V
+-----------+
| RESUMED |
+-----------+
注意：tryResume 和 trySuspend 最多可以被调用一次，哪个先被调用就进入哪个状态
*/
private val _decision = atomic(UNDECIDED)
// 初始化可取消性
public override fun initCancellability() {
val handle = installParentHandle()
?: return // 如果没有父任务，下面的代码就不用执行了
// 如果父任务已完成，则直接释放资源
if (isCompleted) {
handle.dispose()
parentHandle = NonDisposableHandle
}
}
// 安装父任务句柄
private fun installParentHandle(): DisposableHandle? {
// 没有父任务的话直接返回
val parent = context[Job] ?: return null
// 安装句柄（设置完成回调）
val handle = parent.invokeOnCompletion(
onCancelling = true,
// 创建子续体
handler = ChildContinuation(this).asHandler
)
parentHandle = handle
return handle
}
}
CancellableContinuationImpl 的 initCancellability() 方法会调用 installParentHandle() 方法把自己设为父任务的工作节点，如果父任务已完成的话，就直接让父任务返回的 JobNode 句柄释放资源（把自己从父任务的工作节点列表中移除）。
在 installParentHandle() 方法中，会把自己封装为 ChildContinuation，把调用父任务的 invokeOnCompletion() 方法获取句柄（工作节点）。
ChildContinuation
internal class ChildContinuation(
@JvmField val child: CancellableContinuationImpl<*>
) : JobCancellingNode() {
override fun invoke(cause: Throwable?) {
child.parentCancelled(child.getContinuationCancellationCause(job))
}
}
ChildContinuation 是 JobCancellingNode 的子类，当协程被取消的时候，就会把取消事件和原因传给 JobCancellingNode ，ChildContinuation 收到取消通知，invoke 方法被调用后，就会把异常传给 CancellableContinuationImpl 。
完整时序图
withTimeout+执行时序图.png
参考资料
Kotlin 官方文档取消与超时：https://kotlinlang.org/docs/cancellation-and-timeouts.html
网友评论