不用休眠的 Kotlin 并发：深入对比 delay() 和 sleep()

news2024/10/2 12:32:26

在这里插入图片描述

本文翻译自：
https://blog.shreyaspatil.dev/sleepless-concurrency-delay-vs-threadsleep

毫无疑问，Kotlin 语言中的协程 Coroutine 极大地帮助了开发者更加容易地处理异步编程。该特性中封装的诸多高效 API，可以确保开发者花费更小的精力去完成并发任务。一般来说，开发者了解一下如何使用这些 API 就足够了！

可就 JVM 的角度而言，协程一定程度上减少了*“回调地狱”*的问题，切实地改进了异步处理的编码方式。

相信包括笔者在内的很多开发者常常会好奇协程的背后到底是如何做到的。所以，本文将以 delay() 为切入点，带开发者剖析下协程的背后原理。

目录前瞻：

delay() 干啥用的？
sleep() 呢？
对比 delay() 和 sleep()
剖析 delay() 原理

1. `delay()` 干啥用的？

使用过协程的开发者大概率对 delay() 并不陌生，anyway，先来看下官方针对该函数的描述：

“delay() 用来延迟协程一段时间，但不阻塞线程，并且能在指定的时间后恢复协程的执行。”

来看一段在 task1 执行 2000ms 后执行 task2 的示例代码：

scope.launch {
    doTask1()
    delay(2000)
    doTask2()
}

代码很简单，但需要再次提醒一些关于 delay() 的重要特点：

它不会阻塞当前运行的线程
但它允许其他协程在同线程运行
当延迟的时间到了，协程会被恢复并继续执行

很多开发者常常会将 delay() 和 Java 语言的 sleep() 进行比较。可事实上，这两个函数用作完全不同的场景，只是命名上看起来有点相似而已。。。

2. `sleep()` 呢？

sleep() 则是 Java 语言中标准的多线程处理 API：促使当前执行的线程进入休眠，并持续指定的一段时间。

“该方法一般用来告知 CPU 让出处理时间给 App 的其他线程或者其他 App 的线程。”

如果在协程里使用该函数，它会导致当前运行的线程被阻塞，同时也会导致该线程的其他协程被阻塞，直到指定的阻塞时间完成。

为了解更多的细节，让我们通过示例进一步地对比 sleep() 和 delay() 两者。

3. 对比 delay() 和 sleep()

假使我们想在单线程（就比如 Android 开发里的主线程）里执行并发任务。

看一下如下的代码片段：分别启动两个协程，并各自调用了 1000ms 的 delay() 或 sleep()。

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

比较：

协程的启动时间：
- 调用 delay() 代码里的两个协程在同一时间(05:48:58)执行
- 调用 sleep() 代码里的第 2 个协程相隔了 1s 后执行
协程的结束时间：
- 调用 delay() 代码里的 2 个协程一共花了 1045ms
- 调用 sleep() 代码里的 2 个协程则一共花了 2044ms

这也印证了上面提到的特性差异：delay() 只是挂起协程、同时允许其他协程复用该协程，而 sleep() 则在一段时间内直接阻塞了整个线程。

事实上，delay() 还具备其他神奇的特点，再来看看下面的代码示例：

先定义了一个最大创建 2 个线程的线程池 context 示例
当第 1 个协程启动并执行一个 task 之后，调用 delay() 挂起 1000ms，接着再执行一个 task
在第 1 个协程执行的同时，启动第 2 个协程兵执行耗时 task

通过查看 task 里打印的 log，我们惊讶地发现：delay 函数执行前，它运行在 Duet-1 线程。但当 delay 完成后，它却恢复到了另一个线程：Duet-2。

这是为什么？

原来是因为原线程正在忙于处理第 2 个协程启动的耗时 task，所以 delay 之后它只能恢复到另一个线程。

这就有意思了，看看官方文档的描述。。。

“协程可以挂起一个 thread 并且恢复到另一个 thread！”

既然感受到了 delay() 的魔力，我们就来了解下它背后的工作原理。

4. 剖析 `delay()` 原理

delay() 会先在协程上下文里找到 Delay 的实现，接着执行具体的延时处理。

public suspend fun delay(timeMillis: Long) {
    if (timeMillis <= 0) return

    return suspendCancellableCoroutine sc@ { cont: CancellableContinuation<Unit> ->
        if (timeMillis < Long.MAX_VALUE) {
            cont.context.delay.scheduleResumeAfterDelay(timeMillis, cont)
        }
    }
}

Delay 是 interface 类型，其定义了延时之后调度协程的方法 scheduleResumeAfterDelay() 等。开发者直接调用的 delay()、withTimeout() 正是 Delay 接口提供的支持。

public interface Delay {
    public fun scheduleResumeAfterDelay(timeMillis: Long, continuation: CancellableContinuation<Unit>)

    public fun invokeOnTimeout(timeMillis: Long, block: Runnable, context: CoroutineContext): DisposableHandle =
        DefaultDelay.invokeOnTimeout(timeMillis, block, context)
}

事实上，Delay 接口由运行协程的各 CoroutineDispatcher 实现。

我们知道 CoroutineDispatcher 是抽象类，Dispatchers 类会利用线程相关 API 来实现它。

比如：

Dispatchers.Default、Dispatchers.IO 使用 java.util.concurrent 包下的 Executor API 来实现
Dispatchers.Main 使用 Android 平台上特有的 Handler API 来实现

接着，各 Dispatcher 还需要实现 Delay 接口，主要就是实现 scheduleResumeAfterDelay() ，去返回指定 ms 之后执行协程的 Continuation 实例。

如下是 ExecutorCoroutineDispatcherImpl 类实现该方法的具体代码：

override fun scheduleResumeAfterDelay(timeMillis: Long, continuation: CancellableContinuation<Unit>) {
    (executor as? ScheduledExecutorService)?.scheduleBlock(
        ResumeUndispatchedRunnable(this, continuation),
        continuation.context,
        timeMillis
    )
    // Other implementation 
}

可以看到：它借助了 Java 包 ScheduledExecutorService 的 schedule() 来调度了 Continuation 的恢复。

我们再来看下 Android 平台 Dispatcher 即 HandlerDispatcher 又是如何实现的该方法。

override fun scheduleResumeAfterDelay(timeMillis: Long, continuation: CancellableContinuation<Unit>) {
    val block = Runnable {
        with(continuation) { resumeUndispatched(Unit) }
    }
    handler.postDelayed(block, timeMillis.coerceAtMost(MAX_DELAY))
    // Other implementation 
}