Kotlin 流程 - 如何处理取消

如何解决Kotlin 流程 - 如何处理取消

我正在学习 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. 由于没有更多消息，将不会再次调用发射，因此即使我们收到了我们想要的所有消息，循环仍将继续在不必要的轮询上浪费资源。

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

解决方法

这里要记住的重要一点是流是“冷的”，至少在它们的简单形式中是这样。这意味着流不能做任何工作，除非你正在积极地从它那里消费数据。冷流没有与之关联的协程。您可以从 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，它将永远不会将控制权返回给收集器。这与在普通代码中编写无限循环相同，对于流来说并不是特别特殊。

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

,

是的，当 emit 取消流时，CancellationException 会抛出 take。

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