我刚刚开始熟悉 Kotlin 流程。
为此,我使用它们来解析二进制文件的内容,我将使用以下流程进行模拟:
fun testFlow() = flow {
println("Starting loop")
try {
for (i in 0..5) {
emit(i)
delay(100)
}
println("Loop has finished")
}
finally {
println("Finally")
}
}
现在,我基本上需要多次文件内容来提取不同的信息集。但是,我不想读取文件两次,而只想读取一次。
由于似乎没有克隆/复制流程的内置机制,我开发了以下辅助函数:
interface MultiConsumeBlock<T> {
suspend fun subscribe(): Flow<T>
}
suspend fun <T> Flow<T>.multiConsume(capacity: Int = DEFAULT_CONCURRENCY, scope: CoroutineScope? = null, block: suspend MultiConsumeBlock<T>.() -> Unit) {
val channel = buffer(capacity).broadcastIn(scope ?: CoroutineScope(coroutineContext))
val context = object : MultiConsumeBlock<T> {
override suspend fun subscribe(): Flow<T> {
val subscription = channel.openSubscription()
return flow { emitAll(subscription) }
}
}
try {
block(context)
} finally {
channel.cancel()
}
}
然后我像这样使用它(考虑与文件的类比:flowa
获取每条记录,b
仅流动前 3 条记录(=“file header”)并c
在 header 之后流动所有内容):
fun main() = runBlocking {
val src = testFlow()
src.multiConsume {
val a = subscribe().map { it }
val b = subscribe().drop(3).map{ it + it}
val c = subscribe().take(3).map{ it * it}
mapOf("A" to a, "B" to b, "C" to c).map { task -> launch { task.value.collect{ println("${task.key}: $it")} } }.toList().joinAll()
}
}
输出:
Starting loop
A: 0
C: 1
A: 1
C: 2
A: 4
C: 3
A: 9
C: 4
A: 16
C: 5
B: 10
C: 6
B: 12
C: 7
B: 14
C: 8
B: 16
C: 9
B: 18
C: 10
B: 20
C: 11
Loop has finished
Finally
到目前为止看起来不错。但是,我不确定在这方面我是否正确使用了 Kotlin 的流程。
我是否为死锁、错过的异常等敞开心扉?
Flow 接口的所有实现都必须遵守下面详细描述的两个关键属性:
- 上下文保存。
- 异常透明度。
但我不确定我的实施是否属于这种情况,或者我是否遗漏了什么。
或者也许有更好的方法?