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)
}

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

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

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

Answer 1

这里要记住的重要一点是，流是“冷的”，至少在它们的简单形式中是这样。这意味着一个流不能做任何工作，除非你正在积极地使用它的数据。冷流没有与之关联的协程。您可以通过this blog post by Roman Elizarov了解更多信息。

当您在流上调用collect 时，控制会从收集器转移到流。这就是使流程能够工作的原因。收集器有效地执行流内的代码。当流调用emit 时，控制权转移回收集器。如果您熟悉 Kotlin 的 sequence builder，您可以非常相似地想到流程。

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

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

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

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

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

Answer 2

是的，emit 将在 take 取消流程时抛出 CancellationException。

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