以下代碼安全嗎？為什么？
val flow: Flow<String> = ...
val allStrings = mutableListOf<String>()
var sum = 0
flow.transform {
allStrings = it
emit(it.toInt())
}.collect {
sum = it
}
以下測驗演示collect {}了從不同執行緒呼叫的主體：
val ctx = newFixedThreadPoolContext(32, "my-context")
runBlocking(ctx) {
val f = flow<Int> {
(1 .. 1000).forEach {
emit(it)
}
}
var t: Thread? = null
f.collect {
delay(1)
// this requirement will fail
if (t == null) t = Thread.currentThread() else require(t == Thread.currentThread())
}
}
另一個測驗出版物：
fun main(args: Array<String>) {
val ctx = newFixedThreadPoolContext(32, "my-context")
runBlocking(ctx) {
val f = flow<Int> {
(1 .. 1000000).forEach {
emit(it)
}
}
var c = 0
f.transform {
c = 1
boo()
c = 1
emit(it)
c = 1
}.collect {
c = 1
boo()
c = 1
}
println(c) // prints 5_000_000
}
}
suspend fun boo() {
withContext(Dispatchers.IO) {
}
}
因此，似乎kotlin flow 確保了協程呼叫之間的發布，但這是有意的（甚至記錄在案的）還是實作的副作用？
uj5u.com熱心網友回復：
是的，代碼是安全的。默認情況下，流是順序的。
除非使用對多個流進行操作的特殊運算子，否則每個單獨的流集合都是按順序執行的。該集合直接在呼叫終端運算子 (
collect) 的協程中作業。默認情況下不會啟動新的協程。每個發出的值由上游到下游的所有中間算子處理，然后傳遞給終端算子。
Kotlin Flows 基于掛起函式，它們是完全順序的，但不是單執行緒的。
因此，似乎 kotlin flow 確保了協程呼叫之間的發布，但這是有意的（甚至記錄在案的）還是實作的副作用？
這里說：
流是一種可以順序發出多個值的型別
據此，我了解在處理先前的值之前不會收集新值，無論涉及多少執行緒。
正如 Roman 在他的評論中提到的，這是一篇關于順序執行的好文章。來自那里的一個很好的報價：
盡管 Kotlin 中的協程可以在多個執行緒上執行，但從可變狀態的角度來看，它就像一個執行緒。同一個協程中的兩個動作不能同時發生。就像執行緒一樣，您應該避免在協程之間共享可變狀態，否則您將不得不自己擔心同步。避免共享可變狀態。將每個可變物件限制為單個執行緒或單個協程并睡得很好。
這適用于Flows，因為集合Flow直接在呼叫終端運算子的協程中作業。默認情況下不會啟動新的協程。
轉載請註明出處，本文鏈接：https://www.uj5u.com/net/422767.html
標籤：
上一篇：web::block()因“`NonNull<pq_sys::pg_conn>`無法在執行緒之間安全共享”而失敗
下一篇：為什么使用gevent.joinall()而不是pool.imap_unordered()來運行Greenlets？