我正在处理通过回调到达的热事件流。在“下游”,我想将其拆分成多个流并进行处理。这些事件是按顺序从单个线程(我无法控制,因此不认为可以在此处使用协同程序)中依次到达的。
在这里使用什么结构是正确的?
我可以很容易地创建一个 Flow,使用 callbackFlow 和 sendBlocking,但语义似乎不匹配,因为 Flow 不是冷的。将 Flow 拆分为多个下游流(取决于事件内容)的最佳方法是什么?或者我应该使用 Channel?它与我的源的“热度”相匹配,但整个轮询下游似乎有些偏差(在这种基本上同步的情况下),并且许多方法似乎已被淘汰,而改用 Flow。
我可以通过只使用“全部回调”来完成所有这些工作,但这会创建比我希望的更紧密的耦合。有什么好的想法吗?
编辑:
最终我得到了这个,看起来可以工作:
在这里使用什么结构是正确的?
我可以很容易地创建一个 Flow,使用 callbackFlow 和 sendBlocking,但语义似乎不匹配,因为 Flow 不是冷的。将 Flow 拆分为多个下游流(取决于事件内容)的最佳方法是什么?或者我应该使用 Channel?它与我的源的“热度”相匹配,但整个轮询下游似乎有些偏差(在这种基本上同步的情况下),并且许多方法似乎已被淘汰,而改用 Flow。
我可以通过只使用“全部回调”来完成所有这些工作,但这会创建比我希望的更紧密的耦合。有什么好的想法吗?
编辑:
最终我得到了这个,看起来可以工作:
fun testFlow() {
runBlocking {
val original = flowOf("aap", "noot", "mies", "wim", "zus","jet","weide","does")
val broadcast = original.broadcastIn(this)
val flow1 = broadcast.openSubscription().receiveAsFlow().filter { it.length == 4 }
val flow2 = broadcast.openSubscription().receiveAsFlow().filter { it.length == 3 }
flow1.collect { it -> println("Four letter: ${it}") }
flow2.collect { it -> println("Three letter: ${it}") }
}
}
receiveAsFlow()
不会自动取消订阅通道,因此在这种情况下应该优先选择consumeAsFlow()
。话虽如此,openSubscription().consumeAsFlow()
会立即创建2个热流,而使用asFlow()
实际上将“子流”保持为冷态,只有在收集时才会创建订阅。 - Joffrey