1

我正在学习 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() 吗？

4

2 回答 2

1

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

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

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

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

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

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

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

于 2021-07-26T11:56:50.460 回答
0

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

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

于 2021-07-23T12:06:44.410 回答