Kotlin: 13. Flow : 스트림 기반 비동기 처리의 핵심

Kotlin Flow: 스트림 기반 비동기 처리의 핵심

Kotlin Flow는 Coroutine을 기반으로 한 스트림 기반 비동기 처리 도구입니다. 데이터를 순차적으로 발행하고 처리하는 스트림 형태로, 비동기적인 데이터 흐름을 효율적으로 관리하고 처리할 수 있도록 돕습니다. Flow는 여러 값을 순차적으로 처리하는 데 특화되어 있으며, 네트워크 통신, 이벤트 처리, 데이터 스트림 처리 등 다양한 분야에서 사용됩니다.

1. Flow의 핵심 개념

• Cold Flow: 발행자가 소비자(콜렉터)가 존재할 때만 데이터를 발행하는 스트림입니다. 콜렉터가 없으면 데이터는 발행되지 않습니다.
• Hot Flow: 발행자가 소비자가 존재하든 말든 데이터를 발행하는 스트림입니다. 소비자가 늦게 연결되더라도 이전에 발행된 데이터를 받을 수 있습니다.
• 발행자 (Emitter): 데이터를 스트림에 발행하는 객체입니다. flow 빌더를 사용하여 구현합니다.
• 콜렉터: 스트림에서 데이터를 수집하는 객체입니다. collect 함수를 사용하여 구현합니다.
• 백프레셔: 다운스트림의 처리 속도에 따라 업스트림의 데이터 발행 속도를 조절하는 기능입니다. 메모리 누수를 방지하고 효율적인 데이터 처리를 가능하게 합니다.
• 취소 가능성: 콜렉션 중 취소를 지원하여 리소스 낭비를 방지합니다.
• 연산자 (Operator): 스트림을 변환하고 처리하는 기능입니다. 다양한 연산자를 사용하여 스트림을 조작하고 원하는 데이터를 추출할 수 있습니다.

2. Flow 생성 및 사용 예시

      import kotlinx.coroutines.flow.*
import kotlinx.coroutines.launch

fun main() {
    GlobalScope.launch {
        // 1. Flow 생성
        val flow = flow {
            emit("Hello")
            delay(1000) // 1초 지연
            emit("World!")
        }

        // 2. Flow 콜렉션
        flow.collect { value ->
            println(value) 
        }
    }
}

위 코드에서 flow 빌더를 사용하여 "Hello"와 "World!"를 발행하는 Flow를 생성합니다. collect 함수를 사용하여 Flow에서 데이터를 수집하고 출력합니다.

3. Flow 연산자: 다양한 스트림 처리 기능

Flow는 스트림을 변환하고 처리하기 위한 다양한 연산자를 제공합니다. 몇 가지 주요 연산자는 다음과 같습니다.

• map: 각 데이터를 변환합니다.

      flow.map { it.uppercase() } // 모든 데이터를 대문자로 변환
    

• filter: 조건에 맞는 데이터만 필터링합니다.

      flow.filter { it.length > 5 } // 길이가 5 이상인 데이터만 필터링
    

• flatMapConcat: 각 데이터를 새로운 스트림으로 변환하고 순차적으로 합쳐 새로운 스트림을 생성합니다.

      flow.flatMapConcat { value ->
    flow { emit(value.toInt()) } // 각 문자열을 정수로 변환
}
    

• onEach: 각 데이터를 처리하는 추가 작업을 수행합니다.

      flow.onEach { println("Processing: $it") } // 각 데이터 처리 시 로그 출력
    

• distinctUntilChanged: 이전 데이터와 다른 데이터만 발행합니다.

      flow.distinctUntilChanged() // 이전 데이터와 같은 데이터는 발행하지 않음
    

• reduce: 스트림의 데이터를 하나로 축소합니다.

      flow.reduce { accumulator, value -> accumulator + value } // 모든 데이터를 합산
    

• catch: 에러 처리를 위한 연산자입니다.

      flow.catch { error -> println("Error: $error") } // 에러 발생 시 로그 출력
    

4. Flow의 장점

• 비동기 처리: Coroutine을 기반으로 하므로 비동기 작업을 효율적으로 처리할 수 있습니다.
• 스트림 기반 처리: 데이터를 스트림 형태로 처리하여 효율성을 높입니다.
• 백프레셔 지원: 다운스트림의 처리 속도에 따라 업스트림의 데이터 발행 속도를 조절하여 메모리 누수를 방지합니다.
• 취소 가능성: 콜렉션 중 취소를 지원하여 리소스 낭비를 방지합니다.
• 조합 연산자: 다양한 조합 연산자를 제공하여 스트림을 효율적으로 변환 및 처리할 수 있습니다.

 

Kotlin Flow vs Coroutine: 비동기 처리의 두 도구 비교

Kotlin에서 비동기 처리를 위한 두 가지 강력한 도구인 Flow와 Coroutine은 서로 밀접하게 관련되어 있지만, 각자의 특징과 적합한 용도가 있습니다.

Coroutine은 비동기 작업을 쉽게 처리할 수 있도록 돕는 기본적인 도구입니다. 반면 Flow는 Coroutine을 기반으로 하여 데이터를 스트림 형태로 처리하는 데 특화된 도구입니다.

1. 핵심 차이점:

특징	Coroutine	Flow
데이터 처리	단일 값	여러 값 (스트림)
적용 분야	단일 비동기 작업	데이터 스트림 처리, 네트워크 통신, 이벤트 처리 등
장점	단순하고 직관적인 비동기 작업 처리	스트림 기반 처리를 통한 효율성 및 유연성

2. 각 도구의 적합한 용도:

• Coroutine: 단일 비동기 작업을 처리하는 데 적합합니다. 예를 들어, 네트워크 요청을 보내고 응답을 처리하는 경우 Coroutine을 사용하여 간단하게 비동기 처리를 구현할 수 있습니다.

  

  Coroutine : 목적지를 향해 가는 자동차 (단일)

  
  
• Flow: 데이터를 스트림 형태로 처리해야 할 때 유용합니다. 예를 들어, 파일에서 데이터를 읽어들이거나 웹소켓을 통해 데이터를 수신하는 경우 Flow를 사용하여 데이터를 효율적으로 처리할 수 있습니다.

Flow : 컨베이어 벨트와 같이 많은 양을 처리 (스트림)

3. 비교 예시:

네트워크 데이터 가져오기

• Coroutine:

      import kotlinx.coroutines.*
import okhttp3.*

fun main() {
    GlobalScope.launch {
        val client = OkHttpClient()
        val url = "https://api.example.com/data"
        val request = Request.Builder().url(url).build()

        val response = client.newCall(request).execute()
        if (response.isSuccessful) {
            val data = response.body!!.string()
            println(data)
        } else {
            println("Request failed: ${response.code}")
        }
    }
}
    

• Flow:

      import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import okhttp3.*

fun main() {
    GlobalScope.launch {
        val client = OkHttpClient()
        val url = "https://api.example.com/data"

        flow {
            val response = client.newCall(Request.Builder().url(url).build()).execute()
            if (response.isSuccessful) {
                val data = response.body!!.string()
                emit(data) // 데이터를 스트림에 발행
            } else {
                throw IOException("Request failed: ${response.code}")
            }
        }.catch { 
            println("Error: ${it.message}")
        }.collect {
            println(it) 
        }
    }
}
    

두 예제 모두 네트워크 요청을 보내고 응답을 처리하는 코드입니다. Coroutine은 단일 요청을 처리하는 반면, Flow는 응답 데이터를 스트림 형태로 처리합니다.

Kotlin Flow 예제

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import android.widget.TextView
import java.util.concurrent.TimeUnit

// 가상 서버 데이터
data class ServerData(val message: String)

// 가상 서버 클래스 (실제로는 네트워크 통신 코드로 대체)
class MockServer {
    fun getRealTimeData(): Flow<ServerData> {
        return flow {
            while (true) {
                delay(2000) // 2초 간격으로 데이터 발행
                emit(ServerData("새로운 데이터: ${System.currentTimeMillis()}"))
            }
        }
    }
}

// UI 업데이트 함수 (실제로는 UI 프레임워크에 맞게 수정)
fun updateUI(message: String, textView: TextView) {
    textView.text = message
}

fun main() {
    // UI 요소 설정 (실제 코드에서는 UI 프레임워크에서 가져옴)
    val textView = TextView(this) 

    // Flow를 사용하여 서버에서 데이터 수신
    GlobalScope.launch {
        val server = MockServer()
        server.getRealTimeData().collect { data ->
            updateUI(data.message, textView) // UI 업데이트
        }
    }
}