როგორ შეიძლება გადართოთ ნაკადები Flow-ში? რომელ ნაკადზე მუშაობს Flow?

Үздіксіз, Flow іске қосылған жіпте жұмыс істейді. Жіптерді ауыстыру үшін flowOn операторын қолдануға болады.

import kotlinx.coroutines.CoroutineDispatcher // CoroutineDispatcher үшін
import kotlinx.coroutines.Dispatchers // Dispatchers үшін
import kotlinx.coroutines.flow.Flow // Flow үшін
import kotlinx.coroutines.flow.flow // flow үшін
import kotlinx.coroutines.flow.flowOn // flowOn үшін
import kotlinx.coroutines.runBlocking // runBlocking үшін

// Мысал функция, ол Flow қайтарады
fun simpleFlow(): Flow<Int> = flow {
    logThread("Flow басталды") // Жіпті логтау, онда flow құрастырушы іске қосылады
    for (i in 1..3) {
        emit(i) // Мәндерді шығару
    }
}

fun main() = runBlocking {
    simpleFlow()
        .flowOn(Dispatchers.IO) // Жоғарыдан төменгі (эмиттерлер) Dispatchers.IO-ға ауыстырылды
        .collect { value ->
            logThread("Жиналған $value") // Жіпті логтау, онда мәндер жиналады
        }
}

// Жіпті логтау үшін көмектесетін функция
fun logThread(msg: String) {
    println("[$msg] [${Thread.currentThread().name}]")
}

flowOn операторы оның алдындағы операторлар орындалатын жіпті әсер етеді (оның ішінде flow құрастырушы). flowOn-нан кейінгі операторлар көрсетілген жіпте орындалады. Егер бірнеше flowOn болса, әрқайсысы өз алдындағы бөлікті әсер етеді.

Көптеген flowOn қолдану мысалы:

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

fun simpleFlowWithMultipleFlowOn(): Flow<String> = flow {
    logThread("Flow құрастырушы басталды") // Бірінші көрсетілген `flowOn`-да жұмыс істейді
    emit("A")
    emit("B")
}.map {
    logThread("Карталау $it") // Екінші көрсетілген `flowOn`-да жұмыс істейді
    it.toLowerCase()
}.flowOn(Dispatchers.Default) // Екінші `flowOn` map және flow құрастырушы үстінде әсер етеді
.filter {
    logThread("Фильтрация $it") // `collect` шақырылған жіпте жұмыс істейді (әдетте main/runBlocking)
    true
}.flowOn(Dispatchers.IO) // Бірінші `flowOn` map және flow үстінде әсер етеді

fun main() = runBlocking {
    simpleFlowWithMultipleFlowOn().collect { value ->
        logThread("Жинау $value") // runBlocking (main) жіпте жұмыс істейді
    }
}

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

Бұл мысалда:

• flow құрастырушы және map Dispatchers.Default-да жұмыс істейді (соңғы flowOn-ның себебі)
• filter collect шақырылған жіпте жұмыс істейді, яғни runBlocking орындалатын жіпте (әдетте Main)

Маңызды түсіну керек, flowOn аралық буфер құрады және ол "ашық" ауыстыру емес.