Kotlin Academy · 강의

flowOn과 buffer: 컨텍스트와 백프레셔

flowOn으로 방출 컨텍스트를 변경하고 buffer로 백프레셔에 대비해 방출 값을 버퍼링해 보세요.

레슨 4/413개 단계

flowOn과 buffer: 컨텍스트와 백프레셔은(는) CoddyKit의 무료 Kotlin Academy 강의입니다. 이것은 4개 중 4번째 강의입니다. 아래에서 전체 강의를 무료로 읽을 수 있으며, 내장 코드 에디터와 24/7 AI 튜터와 함께 브라우저에서 직접 실습할 수 있습니다. 이 강의는 Kotlin Academy 학습 경로의 일부이며, 진행 상황이 웹과 CoddyKit 앱에 동기화됩니다. Kotlin Academy 강의에는 총 4개의 강의가 포함되어 있습니다.

플로우 컨텍스트

기본적으로 플로우는 collect를 호출하는 코루틴의 컨텍스트에서 실행됩니다. 변경하지 않는 한 생산자와 소비자의 디스패처는 동일합니다.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    flow {
        println("emit on: ${Thread.currentThread().name}")
        emit(1)
    }.collect {
        println("collect on: ${Thread.currentThread().name}")
    }
}

flowOn으로 상류 컨텍스트 변경하기

flowOn(dispatcher)은 지정한 디스패처에서 상류 플로우(체인에서 그 연산자보다 위에 있는 모든 부분)를 실행하고, 수집은 호출자의 디스패처에서 계속 실행합니다.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    flow {
        println("emit: ${Thread.currentThread().name}")
        emit(1)
    }.map {
        println("map: ${Thread.currentThread().name}")
        it * 2
    }.flowOn(Dispatchers.Default)  // above runs on Default
    .collect {
        println("collect: ${Thread.currentThread().name}")
    }
}

체인에서 flowOn 여러 번 사용하기

flowOn은 여러 번 사용할 수 있습니다. 각 flowOn은 이전 flowOn까지 바로 위에 있는 연산자에 영향을 줍니다.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    flow { emit(1) }
        .map { it + 1 }.flowOn(Dispatchers.IO)      // map runs on IO
        .map { it * 2 }.flowOn(Dispatchers.Default)  // this map runs on Default
        .collect { println("Result: $it") }           // collect on Main (runBlocking)
}

백프레셔 문제

생산자가 수집기에서 처리할 수 있는 속도보다 빠르게 값을 방출하면 값이 대기열에 쌓입니다. 버퍼링이 없으면 생산자가 기다리게 됩니다.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    val time = System.currentTimeMillis()
    flow {
        repeat(3) { i ->
            delay(100)  // fast producer
            emit(i)
        }
    }.collect {
        delay(300)      // slow consumer
        println("Got $it in ${System.currentTimeMillis() - time}ms")
    }
}

buffer() 연산자

buffer()는 별도의 코루틴에서 생산자와 소비자를 동시에 실행하고, 방출된 값을 채널에 버퍼링합니다. 생산자는 소비자를 기다리지 않습니다.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    val time = System.currentTimeMillis()
    flow {
        repeat(3) { i -> delay(100); emit(i) }
    }.buffer()  // producer and consumer run concurrently
    .collect {
        delay(300)
        println("Got $it in ${System.currentTimeMillis() - time}ms")
    }
}

buffer 용량

buffer(capacity)는 채널 버퍼의 크기를 설정합니다. 버퍼가 가득 차면 생산자가 일시 중단됩니다(백프레셔). 기본 용량은 64입니다.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.channels.Channel
fun main() = runBlocking {
    flow { repeat(5) { emit(it) } }
        .buffer(Channel.RENDEZVOUS)    // 0: producer waits
        // .buffer(Channel.BUFFERED)   // default: 64
        // .buffer(Channel.UNLIMITED)  // unbounded
        .collect { delay(50); println(it) }
}

느린 수집기에 collectLatest 사용하기

collectLatest는 새 값이 도착하면 현재 수집 블록을 취소하고 최신 값으로 다시 시작합니다.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    flow {
        emit(1); delay(50)
        emit(2); delay(50)
        emit(3)
    }.collectLatest { value ->
        println("Processing $value")
        delay(100)  // gets cancelled if new value arrives
        println("Done $value")  // only prints for last value
    }
}

flowOn + buffer 패턴

flowOn과 buffer를 결합하십시오. IO(네트워크/디스크)에서 생산자를 실행하고, 결과를 버퍼링한 다음, Main에서 수집합니다.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun fetchItems(): Flow<String> = flow {
    repeat(3) { i ->
        delay(100)  // simulate IO
        emit("item-$i")
    }
}.flowOn(Dispatchers.IO).buffer(10)
fun main() = runBlocking {
    fetchItems().collect { println("UI: $it") }
}

동시 생산자에 channelFlow 사용하기

channelFlow는 채널을 기반으로 하는 플로우를 생성하여 빌더 내부에서 여러 코루틴이 동시에 값을 방출할 수 있게 합니다.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun concurrentFlow(): Flow<Int> = channelFlow {
    launch { send(1) }
    launch { send(2) }
    launch { send(3) }
}
fun main() = runBlocking {
    concurrentFlow().collect { println(it) }
}

알맞은 전략 선택하기

요약하면, 컨텍스트 전환에는 flowOn, 처리량 향상에는 buffer, 최신 UI 값만 유지하려면 conflate, 오래된 처리를 취소하려면 collectLatest를 사용하십시오.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
// Guidelines:
// CPU-heavy production  -> flowOn(Dispatchers.Default)
// IO-heavy production   -> flowOn(Dispatchers.IO)
// Slow consumer         -> buffer()
// UI state updates      -> conflate() or StateFlow
// Search/autocomplete   -> collectLatest or flatMapLatest
fun main() = runBlocking { println("Choose the right strategy!") }

빠른 확인

플로우 체인에서 flowOn은 무엇에 영향을 줍니까?

인터랙티브 레슨에서 답해보세요

복습

flowOn은 상류 작업을 다른 디스패처로 이동합니다. buffer는 처리량을 높이기 위해 생산자와 소비자를 분리합니다. conflate는 중간 값을 버립니다. collectLatest는 새 값이 도착하면 느린 처리를 취소합니다.

무료로 시작

AI 튜터와 함께 Kotlin을(를) 배우세요 — 무료

브라우저에서 실제 코드를 작성하고 실행하며, 24/7 AI 튜터로부터 즉각적인 도움을 받고, 웹이나 앱에서 중단한 부분부터 계속 학습하세요.

코스: 51
레슨: 203

무료로 학습 시작 전체 코스 보기

자주 묻는 질문