kotlin - 延迟功能在Kotlin中如何工作而不阻塞当前线程？

Question

过去几天，我正在学习协程，您的大多数概念都很清楚，但是我不理解延迟功能的实现。

延迟时间过后，延迟功能如何恢复协程？对于一个简单的程序，只有一个主线程，为了在延迟的时间后恢复协程，我认为应该有另一个计时器线程来处理所有延迟的调用并在以后调用它们。是真的吗有人可以解释延迟功能的实现细节吗？

Answer 1

TL; DR;

使用runBlocking时，延迟会在内部包装并在同一线程上运行；使用其他任何调度程序时，它会挂起并通过恢复事件循环线程的继续来恢复。查看下面的详细答案以了解内部原理。

长答案:

@Francesc答案正确指出，但有点抽象，仍然不能解释延迟在内部的实际作用。

因此，正如他指出的延迟功能:

public suspend fun delay(timeMillis: Long) {
    if (timeMillis <= 0) return // don't delay
    return suspendCancellableCoroutine sc@ { cont: CancellableContinuation<Unit> ->
        cont.context.delay.scheduleResumeAfterDelay(timeMillis, cont)
    }
}

它的作用是“在lambda 内部运行该块后，在suspend函数中获取当前的延续实例，并暂停当前正在运行的协程 ”

因此，这行cont.context.delay.scheduleResumeAfterDelay(timeMillis, cont)将要执行，然后当前的协程被挂起，即释放它所坚持的当前线程。
cont.context.delay指向

internal val CoroutineContext.delay: Delay get() = get(ContinuationInterceptor) as? Delay ?: DefaultDelay

这表示如果ContinuationInterceptor是Delay的实现，则返回该值，否则返回DefaultDelay，它是internal actual val DefaultDelay: Delay = DefaultExecutor，一个DefaultExecutor，它是internal actual object DefaultExecutor : EventLoopImplBase(), Runnable {...}的EventLoop实现，并有自己的线程可以运行。

注意:当协程位于runBlocking块中时，ContinuationInterceptor是Delay的实现，以确保延迟在同一线程上运行，否则就不在同一线程上运行。检查this snippet以查看结果。

现在我找不到由runBlocking创建的Delay的实现，因为internal expect fun createEventLoop(): EventLoop是一个期望函数，它是从外部而不是源实现的。但是DefaultDelay实现如下

public override fun scheduleResumeAfterDelay(timeMillis: Long, continuation: CancellableContinuation<Unit>) {
    val timeNanos = delayToNanos(timeMillis)
    if (timeNanos < MAX_DELAY_NS) {
        val now = nanoTime()
        DelayedResumeTask(now + timeNanos, continuation).also { task ->
            continuation.disposeOnCancellation(task)
            schedule(now, task)
        }
    }
}

这就是scheduleResumeAfterDelay的实现方式，它通过延迟传递连续性来创建DelayedResumeTask，然后调用schedule(now, task)，后者调用scheduleImpl(now, delayedTask)，最后调用delayedTask.scheduleTask(now, delayedQueue, this)，并在对象中传递delayedQueue

@Synchronized
fun scheduleTask(now: Long, delayed: DelayedTaskQueue, eventLoop: EventLoopImplBase): Int {
    if (_heap === kotlinx.coroutines.DISPOSED_TASK) return SCHEDULE_DISPOSED // don't add -- was already disposed
    delayed.addLastIf(this) { firstTask ->
        if (eventLoop.isCompleted) return SCHEDULE_COMPLETED // non-local return from scheduleTask
        /**
         * We are about to add new task and we have to make sure that [DelayedTaskQueue]
         * invariant is maintained. The code in this lambda is additionally executed under
         * the lock of [DelayedTaskQueue] and working with [DelayedTaskQueue.timeNow] here is thread-safe.
         */
        if (firstTask == null) {
            /**
             * When adding the first delayed task we simply update queue's [DelayedTaskQueue.timeNow] to
             * the current now time even if that means "going backwards in time". This makes the structure
             * self-correcting in spite of wild jumps in `nanoTime()` measurements once all delayed tasks
             * are removed from the delayed queue for execution.
             */
            delayed.timeNow = now
        } else {
            /**
             * Carefully update [DelayedTaskQueue.timeNow] so that it does not sweep past first's tasks time
             * and only goes forward in time. We cannot let it go backwards in time or invariant can be
             * violated for tasks that were already scheduled.
             */
            val firstTime = firstTask.nanoTime
            // compute min(now, firstTime) using a wrap-safe check
            val minTime = if (firstTime - now >= 0) now else firstTime
            // update timeNow only when going forward in time
            if (minTime - delayed.timeNow > 0) delayed.timeNow = minTime
        }
        /**
         * Here [DelayedTaskQueue.timeNow] was already modified and we have to double-check that newly added
         * task does not violate [DelayedTaskQueue] invariant because of that. Note also that this scheduleTask
         * function can be called to reschedule from one queue to another and this might be another reason
         * where new task's time might now violate invariant.
         * We correct invariant violation (if any) by simply changing this task's time to now.
         */
        if (nanoTime - delayed.timeNow < 0) nanoTime = delayed.timeNow
        true
    }
    return SCHEDULE_OK
}

最后，它将任务与当前时间一起设置为DelayedTaskQueue。

// Inside DefaultExecutor
override fun run() {
    ThreadLocalEventLoop.setEventLoop(this)
    registerTimeLoopThread()
    try {
        var shutdownNanos = Long.MAX_VALUE
        if (!DefaultExecutor.notifyStartup()) return
        while (true) {
            Thread.interrupted() // just reset interruption flag
            var parkNanos = DefaultExecutor.processNextEvent() /* Notice here, it calls the processNextEvent */
            if (parkNanos == Long.MAX_VALUE) {
                // nothing to do, initialize shutdown timeout
                if (shutdownNanos == Long.MAX_VALUE) {
                    val now = nanoTime()
                    if (shutdownNanos == Long.MAX_VALUE) shutdownNanos = now + DefaultExecutor.KEEP_ALIVE_NANOS
                    val tillShutdown = shutdownNanos - now
                    if (tillShutdown <= 0) return // shut thread down
                    parkNanos = parkNanos.coerceAtMost(tillShutdown)
                } else
                    parkNanos = parkNanos.coerceAtMost(DefaultExecutor.KEEP_ALIVE_NANOS) // limit wait time anyway
            }
            if (parkNanos > 0) {
                // check if shutdown was requested and bail out in this case
                if (DefaultExecutor.isShutdownRequested) return
                parkNanos(this, parkNanos)
            }
        }
    } finally {
        DefaultExecutor._thread = null // this thread is dead
        DefaultExecutor.acknowledgeShutdownIfNeeded()
        unregisterTimeLoopThread()
        // recheck if queues are empty after _thread reference was set to null (!!!)
        if (!DefaultExecutor.isEmpty) DefaultExecutor.thread // recreate thread if it is needed
    }
}

// Called by run inside the run of DefaultExecutor
override fun processNextEvent(): Long {
    // unconfined events take priority
    if (processUnconfinedEvent()) return nextTime
    // queue all delayed tasks that are due to be executed
    val delayed = _delayed.value
    if (delayed != null && !delayed.isEmpty) {
        val now = nanoTime()
        while (true) {
            // make sure that moving from delayed to queue removes from delayed only after it is added to queue
            // to make sure that 'isEmpty' and `nextTime` that check both of them
            // do not transiently report that both delayed and queue are empty during move
            delayed.removeFirstIf {
                if (it.timeToExecute(now)) {
                    enqueueImpl(it)
                } else
                    false
            } ?: break // quit loop when nothing more to remove or enqueueImpl returns false on "isComplete"
        }
    }
    // then process one event from queue
    dequeue()?.run()
    return nextTime
}

然后，internal actual object DefaultExecutor : EventLoopImplBase(), Runnable {...}的事件循环(运行功能)最终通过将任务出队并恢复实际的Continuation来处理任务，如果到达延迟时间，则通过调用delay来暂停该函数。