当 CoroutineScope 被取消时，Coroutine StateFlow 停止发射

Question

我有一个StateFlow 协程，它在我的应用程序的各个部分之间共享。当我cancelCoroutineScope 的下游收集器时，JobCancellationException 被传播到StateFlow，它停止为所有当前和未来的收集器发出值。

StateFlow:

val songsRelay: Flow<List<Song>> by lazy {
    MutableStateFlow<List<Song>?>(null).apply {
        CoroutineScope(Dispatchers.IO)
            .launch { songDataDao.getAll().distinctUntilChanged().collect { value = it } }
    }.filterNotNull()
}

我的代码中的典型“演示者”实现了以下基类：

abstract class BasePresenter<T : Any> : BaseContract.Presenter<T> {

    var view: T? = null

    private val job by lazy {
        Job()
    }

    private val coroutineScope by lazy { CoroutineScope( job + Dispatchers.Main) }

    override fun bindView(view: T) {
        this.view = view
    }

    override fun unbindView() {
        job.cancel()
        view = null
    }

    fun launch(block: suspend CoroutineScope.() -> Unit): Job {
        return coroutineScope.launch(block = block)
    }
}

BasePresenter 实现可能会调用 launch{ songsRelay.collect {...} } 当presenter解绑时，为了防止泄露，我取消了父job。任何时候收集songsRelayStateFlow 的演示者未绑定，StateFlow 本质上以JobCancellationException 终止，并且没有其他收集器/演示者可以从中收集值。

我注意到我可以改为调用job.cancelChildren()，这似乎可行（StateFlow 不能与JobCancellationException 一起完成）。但是后来我想知道如果我不能取消工作本身，那么声明父母job 有什么意义。我可以完全删除 job，然后调用 coroutineScope.coroutineContext.cancelChildren() 达到同样的效果。

如果我只是打电话给job.cancelChildren()，就够了吗？我觉得如果不打电话给coroutineScope.cancel() 或job.cancel()，我可能没有正确或完全清理我已经开始的任务。

我也不明白为什么JobCancellationException 在调用job.cancel() 时会向上传播。 job 不是这里的“父母”吗？为什么取消它会影响我的StateFlow？

Answer 1

更新：

你确定你的songRelay 真的被所有演示者取消了吗？我运行了这个测试并打印了“Song relay completed”，因为onCompletion 也捕获了下游异常。但是 Presenter 2 发出的值 2 就好了，在歌曲中继打印“完成”之后。如果我取消 Presenter 2，则会再次打印“Song relay completed”，并针对 Presenter 2 的作业显示 JobCancellationException。

我确实发现一个流实例如何为每个订阅的收集器发出一次很有趣。关于流量，我没有意识到这一点。

    val songsRelay: Flow<Int> by lazy {
        MutableStateFlow<Int?>(null).apply {
            CoroutineScope(Dispatchers.IO)
                    .launch {
                        flow {
                            emit(1)
                            delay(1000)
                            emit(2)
                            delay(1000)
                            emit(3)
                        }.onCompletion {
                            println("Dao completed")
                        }.collect { value = it }
                    }
        }.filterNotNull()
                .onCompletion { cause ->
                    println("Song relay completed: $cause")
                }
    }

    @Test
    fun test() = runBlocking {
        val job = Job()
        val presenterScope1 = CoroutineScope(job + Dispatchers.Unconfined)
        val presenterScope2 = CoroutineScope(Job() + Dispatchers.Unconfined)

        presenterScope1.launch {
            songsRelay.onCompletion { cause ->
                println("Presenter 1 Completed: $cause")
            }.collect {
                println("Presenter 1 emits: $it")
            }
        }

        presenterScope2.launch {
            songsRelay.collect {
                println("Presenter 2 emits: $it")
            }
        }

        presenterScope1.cancel()

        delay(2000)
        println("Done test")
    }

我认为您需要在 BasePresenter 中使用 SupervisorJob 而不是 Job。一般来说，使用Job 对整个 Presenter 来说都是一个错误，因为一个失败的协程会取消 Presenter 中的所有协程。一般不是你想要的。

Answer 2

好的，所以问题是我在测试时做出了一些错误的假设。 StateFlow 行为正确，取消工作按预期工作。

我在想Presenters、StateFlow 之间会停止发射值，但我实际上是在测试 Presenter 的同一个实例 - 所以它的 Job 已被取消，因此预计不会继续收集流量排放。

我还错误地将onCompletion 的StateFlow 中发出的CancellationException 消息理解为StateFlow 本身已被取消-实际上它只是说下游Collector/Job 已被取消.

我想出了一个更好的BasePresenter 实现，看起来像这样：

abstract class BasePresenter<T : Any> : BaseContract.Presenter<T>, CoroutineScope {

    var view: T? = null

    private var job = Job()

    override val coroutineContext: CoroutineContext
        get() = job + Dispatchers.Main

    override fun bindView(view: T) {
        if (job.isCancelled) {
            job = Job()
        }
        this.view = view
    }

    override fun unbindView() {
        job.cancel()
        view = null
    }
}