取消 Kotlin 流的收集

Question

我有一个具有不同状态的父类，这个父类有一个子类列表，每个子类都有不同的状态。我想收集它们中的每一个并取消达到Terminated 状态的那个。类似的东西：

coroutineScope.launch(Dispatcher.IO) {
   parent.parentState.collect {
      if(it is ParentState.Normal){
         it.children.forEach{ child ->
             coroutineScope.launch(Dispatcher.IO){
                child.childState.collect{
                    if(it is ChildState.Terminated){
                       //when this line executed all the collectors stop until I change the states for each one of them..
                       this.coroutineContext.job.cancel()
                    } else{
                       // Do something else for any other state...
                    }
                }
             }

         }
      }
   }
}

但是当我这样做时，我正在收集的所有孩子都会停止收集，但如果我更改了每个孩子的状态，它会再次开始收集，而在取消其中一个之前情况并非如此。

所以我的问题是为什么在取消其中一位收藏家的工作时会出现这样的行为？

还有没有更好的“反应式”来写这个？

Answer 1

我完全同意您可以使用SupervisorJob 来处理您的问题。

但在我看来，不需要那么多的子作业。一个子作业可以解决你遇到的问题。在您的代码中，将创建一个大小为children Collection 的子协程。虽然协程很轻量级，但我认为是不必要的开销。

您可以在每个Flow collect 之前将List<Flow<T>> 完全转换为Flow<List<T>>。之后，只有转换后的单个Flow可以是collect。

我是这样处理的：

inline fun <reified T> List<Flow<T>>.flattenFlow(): Flow<List<T>> = combine(this@flattenFlow) {
    it.toList()
}

coroutineScope.launch(Dispatcher.IO) {
    parent.parentState.collect {
        if (it is ParentState.Normal) {
            it.flattenFlow().collect {childStateList ->
                childStateList.onEach {childState ->
                    if (childState is ChildState.Terminated) {
                        // Do something when state in Terminated..
                    } else {
                        // Do something else for any other state...
                    }
                }
            }
        }
    }
}

Answer 2

默认情况下，协程作用域在CoroutineContext 中使用Job()，Job() 将取消协程执行或任何正在运行的子程序被取消或抛出异常。

要保持其他子执行保持活动状态，您可以使用特殊的Job，即SupervisorJob()

CoroutineScope(SupervisorJob() + Dispatchers.IO)

另外，让我们为每个孩子创建一个全新的范围

val childScope = CoroutineScope(Dispatchers.IO)
childScope.launch {
    child.childState.collect {
        ....
    }
}

您应该取消那些childScope 而不是coroutineContext。

因此，您的代码将如下所示

val scope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
scope.launch {
   parent.parentState.collect {
      if(it is ParentState.Normal){
         it.children.forEach{ child ->
             val childScope = CoroutineScope(Dispatchers.IO)
             childScope.launch(Dispatcher.IO){
                child.childState.collect{
                    if(it is ChildState.Terminated){
                       //when this line executed all the collectors stop until I change the states for each one of them..
                       childScope.cancel()
                    } else{
                       // Do something else for any other state...
                    }
                }
             }

         }
      }
   }
}