Flow'da nasıl iş parçacıkları arasında geçiş yapılır? Flow hangi iş parçacığında çalışır?

Varsayılan olarak, Flow çalıştırıldığı thread üzerinde çalışır. flowOn operatörü ile thread değiştirebilirsiniz.

import kotlinx.coroutines.CoroutineDispatcher // CoroutineDispatcher için
import kotlinx.coroutines.Dispatchers // Dispatchers için
import kotlinx.coroutines.flow.Flow // Flow için
import kotlinx.coroutines.flow.flow // flow için
import kotlinx.coroutines.flow.flowOn // flowOn için
import kotlinx.coroutines.runBlocking // runBlocking için

// Flow döndüren fonksiyon örneği
fun simpleFlow(): Flow<Int> = flow {
    logThread("Flow başladı") // flow builder'ın çalıştığı thread'in loglanması
    for (i in 1..3) {
        emit(i) // Değerleri yay
    }
}

fun main() = runBlocking {
    simpleFlow()
        .flowOn(Dispatchers.IO) // upstream (yayımcılar) Dispatchers.IO'ye geçiyor
        .collect { value ->
            logThread("Toplanan $value") // değerlerin toplandığı thread'in loglanması
        }
}

// Thread'i loglamak için yardımcı fonksiyon
fun logThread(msg: String) {
    println("[$msg] [${Thread.currentThread().name}]")
}

flowOn operatörü, kendisinden önceki (dahil flow builder) operatörlerin çalıştığı thread'i etkiler. flowOn'den sonra gelen operatörler, flowOn argümanında belirtilen thread üzerinde çalışır. Birden fazla flowOn varsa, her biri kendisinden önceki kısmı etkiler.

Birden fazla flowOn ile örnek:

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking

fun simpleFlowWithMultipleFlowOn(): Flow<String> = flow {
    logThread("Flow oluşturucu başladı") // İlk belirtilen flowOn üzerinde çalışır
    emit("A")
    emit("B")
}.map {
    logThread("Mapping $it") // İkinci belirtilen flowOn üzerinde çalışır
    it.toLowerCase()
}.flowOn(Dispatchers.Default) // İkinci flowOn, map() ve flow oluşturucuyu etkiler
.filter {
    logThread("Filtreleme $it") // collect() çağrıldığı thread üzerinde çalışır (varsayılan main/runBlocking)
    true
}.flowOn(Dispatchers.IO) // Birinci flowOn, map() ve flow oluşturucuyu etkiler

fun main() = runBlocking {
    simpleFlowWithMultipleFlowOn().collect { value ->
        logThread("Toplanan $value") // runBlocking (main) thread üzerinde çalışır
    }
}

fun logThread(msg: String) {
    println("[$msg] [${Thread.currentThread().name}]")
}

Bu örnekte:

• flow oluşturucu ve map, Dispatchers.Default üzerinde çalışır (çünkü en son flowOn onların öncesinde gelir).
• filter, collect çağrıldığı thread üzerinde çalışır (genellikle main veya runBlocking).

Önemli not: flowOn, ara tampon (buffer) oluşturur ve "şeffaf" bir thread değişimi değildir.