Kotlin 流程 - 如何处理取消

Question

我正在学习 kotlin 协程和流程，有一件事对我来说有点晦涩难懂。如果我的常规协程有一个很长的运行循环，我可以使用 isActive 或 EnsureActive 来处理取消。然而，这些并不是为流程定义的，但以下代码正确地完成了流程：

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking
import org.slf4j.LoggerFactory

private val logger = LoggerFactory.getLogger("Main")

fun main() {
    val producer = FlowProducer()
    runBlocking {
        producer
            .produce()
            .take(10)
            .collect {
                logger.info("Received $it")
            }
    }
    logger.info("done")
}


class FlowProducer {
    fun produce() = flow {
        try {
            var counter = 1
            while (true) {
                logger.info("Before emit")
                emit(counter++)
                logger.info("After emit")
            }
        }finally {
            logger.info("Producer has finished")
        }

    }.flowOn(Dispatchers.IO)
}

为什么会出现这样的情况呢？是因为emit是一个可挂起的函数来为我处理取消吗？如果有条件地调用发射该怎么办？例如，该循环实际上轮询来自 Kafka 的记录，并且仅当接收到的记录不为空时才会发出调用。那么我们可以有这样的情况：

1. 我们想要 10 条消息（取 10 条）
2. 实际上kafka主题上只有10条消息
3. 由于没有更多消息，emit 不会再次被调用，因此即使我们收到了我们想要的所有消息，循环仍将继续在不必要的轮询上浪费资源。

不确定我的理解是否正确。在这种情况下我应该在每个循环上调用yield()吗？

Answer 1

这里要记住的重要一点是，流是“冷的”，至少在它们的简单形式中是这样。这意味着除非您主动使用其中的数据，否则流无法执行任何工作。冷流没有与之关联的协程。您可以从Roman Elizarov 的这篇博文中了解更多信息。

当您调用collect流时，控制权将从收集器转移到流。这就是使流程能够工作的原因。收集器正在有效地执行流内的代码。当流调用 时emit，控制权将转移回收集器。如果您熟悉 Kotlin 的序列构建器，您可以非常类似地思考流程。

根据定义，这意味着如果您停止收集流，该流将停止执行任何工作。在您的情况下，因为您使用了take(10)，收集器在收到十个项目后将停止执行流程。因为收集器是实际在流内执行循环的东西，所以当收集器不再收集时，循环不会继续运行。一旦停止使用流程，它就像一个不再被迭代的迭代器。它可以像任何其他对象一样被垃圾收集。

您询问是否应该yield()在流程内调用。在某些情况下这可能很有用，您可以在文档中阅读有关流取消检查的更多信息。就您而言，这是没有必要的，因为：

1. 取消检查仅需要检测何时取消了正在执行流程的协程。当流程自行中止时，例如take(10)发出 10 个项目时，它只是正常终止，而不取消任何协程。
2. 该流程是使用 构建的emit，它已经检查了取消情况。

即使不需要取消检查，仍然可以创建永久运行的流程。如上所述，每次流程调用时，控制权仅转移回收集器emit。因此，如果您的流程无限期地运行而不调用emit，它永远不会将控制权返回给收集器。这与在普通代码中编写无限循环相同，并且对于流来说并不是特别特殊。

请注意，可以创建一个有协程在后台工作的热流。在这种情况下，您需要确保协程正确响应流的取消。

Answer 2

是的，取消流程时emit会抛出。CancellationExceptiontake

您给出的 Kafka 示例实际上会起作用，因为take会在 10 号结束时取消流emit，而不是在 11 号开始时取消流。