如何 cancel/unsubscribe 从协程流

Question

我在尝试过早地从流中取消时注意到一个奇怪的行为。看看下面的例子。

这是一个发出整数值的简单流程

  private fun createFlow() = flow {
        repeat(10000) {
            emit(it)
        }
    }

然后我使用这段代码调用createFlow函数

  CoroutineScope(Dispatchers.Main).launch {
            createFlow().collect {

                Log.i("Main", "$it isActive $isActive")
                if (it == 2) {
                    cancel()
                }
            }
        }

这是打印出来的

0 isActive true
1 isActive true
2 isActive true
3 isActive false
4 isActive false
etc...etc

现在我希望流一旦达到值 2 就应该停止发出整数，但它实际上将 isActive 标志切换为 false 并继续发出而不停止。

当我在发射之间添加延迟时，流的行为符合我的预期。

private fun createFlow() = flow {
    repeat(10000) {
        delay(500) //add a delay
        emit(it)
    }
}

这是再次调用流程后打印出来的内容（这是预期的行为）。

0 isActive true
1 isActive true
2 isActive true

如何才能在不增加延迟的情况下准确地取消指定值的流量发射？

Answer 1

我在 this 相关问题

中遇到了解决方法

我已将我项目中的每个 collect 替换为 safeCollect 函数：

/**
 * Only proceed with the given action if the coroutine has not been cancelled.
 * Necessary because Flow.collect receives items even after coroutine was cancelled
 * https://github.com/Kotlin/kotlinx.coroutines/issues/1265
 */
suspend inline fun <T> Flow<T>.safeCollect(crossinline action: suspend (T) -> Unit) {
  collect {
    coroutineContext.ensureActive()
    action(it)
  }
}

Answer 2

我最近想到了这个

似乎它只有在达到暂停点时才会真正取消，而在您发出的代码中没有这样的点

要解决此问题，请在发射之间添加 yield() 或其他一些暂停函数，如 delay(100)

Answer 3

我想在 1.3.7 version 中补充一点，现在来自 Flow Builder 的排放检查取消状态并且可以正确取消。所以有问题的代码将按预期工作

如何 cancel/unsubscribe 从协程流

How to cancel/unsubscribe from coroutines Flow

android

kotlin-coroutines

kotlinx.coroutines.flow