KOTLIN · 심층 가이드
Kotlin 비동기 완전 정리
구조화된 동시성을 기준으로 코루틴 스코프·취소·예외 전파를 설계하고, Flow 연산자로 비동기 스트림을 조립하는 방법을 21개 주제에 걸쳐 다룹니다.
코루틴의 핵심은 가벼운 스레드가 아니라 생명주기가 트리로 묶인다는 점입니다. 스코프 안에서 시작한 작업은 스코프가 끝날 때까지 부모가 기다리고, 부모가 취소되면 자식도 함께 취소됩니다. 콜백이나 Thread를 직접 다루던 방식에서 넘어오면 이 구조가 제약처럼 느껴지지만, 누수된 비동기 작업을 추적할 일이 사라진다는 점에서 실은 가장 큰 이득입니다. GlobalScope를 쓰고 싶어질 때마다 이 트리를 떠올리면 됩니다.
코루틴 기초 — launch & async로 두 빌더의 차이를 잡고, 곧바로 구조화된 동시성과 CoroutineScope & Dispatcher를 붙여 읽으세요. 이 셋이 한 덩어리입니다. 실패 처리는 코루틴 예외 처리 — SupervisorJob에서 형제까지 죽는 기본 동작과 격리 동작을 비교하면 정리됩니다. 스트림 쪽은 Flow — Cold Stream 다음에 callbackFlow로 콜백 래핑을 보면 리스너 기반 API를 어떻게 흡수하는지 감이 옵니다.
취소는 강제 중단이 아니라 협조적입니다. 코루틴에 취소 신호가 와도, 중단 함수를 하나도 호출하지 않는 계산 루프는 끝까지 돕니다. 이럴 때는 루프 안에서 ensureActive()나 yield()를 불러 줘야 합니다. 반대 방향의 실수도 흔합니다. 취소는 CancellationException으로 전달되는데 이를 광범위한 catch (e: Exception)으로 삼켜 버리면 취소가 무시된 것처럼 보입니다. 정리 코드에서 중단 함수를 써야 한다면 withContext(NonCancellable) 안에 넣습니다.
01코루틴 기초 — launch & async
Kotlin 코루틴의 핵심 빌더. launch는 fire-and-forget, async는 결과를 반환합니다.
Kotlin code
import kotlinx.coroutines.*
fun main() = runBlocking {
// launch — 결과 없이 백그라운드 실행
val job = launch {
delay(100)
println("launch 완료")
}
// async — 결과를 Deferred로 반환
val deferred1 = async { fetchUser(1) }
val deferred2 = async { fetchUser(2) }
// await() — 결과 대기
val user1 = deferred1.await()
val user2 = deferred2.await()
println("$user1, $user2")
// awaitAll — 여러 Deferred 동시 대기
val results = awaitAll(
async { fetchUser(3) },
async { fetchUser(4) },
)
println(results)
job.join() // launch 완료 대기
println("모두 완료")
}
suspend fun fetchUser(id: Int): String {
delay(50) // 네트워크 시뮬레이션 (스레드 차단 없음)
return "User#$id"
}async { } 두 개를 순서대로 await()하면 직렬 실행됩니다. 병렬 실행은 async 먼저 시작하고 나중에 await()하세요.
runBlocking은 현재 스레드를 차단합니다. 프로덕션에서는 최상위 진입점(main 함수)이나 테스트에서만 사용하세요. Android에서는 절대 사용하지 마세요.
02CoroutineScope & Dispatcher
Structured Concurrency의 핵심인 스코프와 실행 컨텍스트(Dispatcher) 이해.
Kotlin code
import kotlinx.coroutines.*
fun main() = runBlocking {
// Dispatchers.Default — CPU 집약적 작업
val cpuResult = withContext(Dispatchers.Default) {
(1..10_000_000).sum()
}
println("CPU 결과: $cpuResult")
// Dispatchers.IO — 파일/네트워크 I/O
val ioResult = withContext(Dispatchers.IO) {
readFileMock()
}
println("IO 결과: $ioResult")
// Dispatchers.Main — UI 업데이트 (Android)
// withContext(Dispatchers.Main) { updateUI() }
// coroutineScope — 자식이 모두 완료될 때까지 대기
coroutineScope {
repeat(3) { i ->
launch { println("작업 $i 완료") }
}
}
println("coroutineScope 완료")
// supervisorScope — 자식 하나 실패해도 나머지 계속
supervisorScope {
launch { throw Exception("실패!") }
launch { delay(100); println("나는 계속 실행") }
}
}
suspend fun readFileMock(): String {
delay(100); return "file content"
}withContext()는 새 코루틴을 만들지 않고 현재 코루틴의 컨텍스트만 바꿉니다. IO → Default 전환 시 오버헤드가 적습니다.
Android에서 GlobalScope 사용은 메모리 누수의 주원인입니다. viewModelScope, lifecycleScope를 사용하세요.
03Flow — Cold Stream
비동기 데이터 스트림을 선언적으로 처리하는 Kotlin Flow. RxJava를 대체합니다.
Kotlin code
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
// flow 빌더 — 구독할 때마다 새로 실행 (cold)
fun temperatureFlow(): Flow<Double> = flow {
while (true) {
val temp = 20.0 + Math.random() * 10
emit(temp)
delay(1000)
}
}
// 변환 연산자
fun main() = runBlocking {
temperatureFlow()
.take(5)
.filter { it > 25.0 }
.map { "⚠️ 고온: ${"%.1f".format(it)}°C" }
.collect { println(it) }
// flowOf — 고정 값 스트림
flowOf(1, 2, 3, 4, 5)
.map { it * it }
.toList()
.also { println(it) }
// catch — 에러 처리
flow<Int> { throw RuntimeException("에러!") }
.catch { e -> emit(-1) }
.collect { println(it) } // -1
// flowOn — 업스트림 Dispatcher 전환
flow { emit(heavyCompute()) }
.flowOn(Dispatchers.Default)
.collect { println(it) }
}
fun heavyCompute() = (1..1000).sum()StateFlow와 SharedFlow는 hot stream입니다. StateFlow는 항상 최신 값을 가지고, SharedFlow는 여러 구독자에게 이벤트를 브로드캐스트합니다.
collect는 flow가 완료될 때까지 현재 코루틴을 중단합니다. 무한 flow라면 take(n)이나 별도 코루틴에서 collect해야 합니다.
04Kotlin Multiplatform 기초
expect/actual로 플랫폼별 코드 분기
Kotlin code
<span class="cm">// Kotlin Multiplatform 기초 예제
// data/prompts.js의 생성 프롬프트로 상세 코드 생성 가능</span>
fun main() { println("Kotlin Multiplatform 기초") }KOTLIN 공식 문서를 함께 참고하세요.
자주 발생하는 실수에 주의하세요.
05Flow — Cold Stream 기초
Kotlin Flow는 비동기 데이터 스트림을 위한 Cold Stream입니다. flow { } 빌더로 값을 순차적으로 방출하며, collect가 호출될 때만 실행됩니다. RxJava보다 간결한 코루틴 기반 리액티브 프로그래밍을 제공합니다.
Kotlin code
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
// Cold Stream: collect 호출 시에만 실행
fun numberFlow(): Flow<Int> = flow {
for (i in 1..5) {
delay(100) // 비동기 작업 시뮬레이션
emit(i) // 값 방출
}
}
fun main() = runBlocking {
// 기본 collect
numberFlow().collect { value ->
println("수신: $value")
}
// 중간 연산자 체이닝
numberFlow()
.filter { it % 2 == 0 }
.map { it * it }
.collect { println("변환 결과: $it") }
// flowOf, asFlow 변환
flowOf("A", "B", "C").collect { print("$it ") }
println()
(1..3).asFlow().collect { print("$it ") }
}Flow는 Cold Stream이므로 collect할 때마다 처음부터 다시 실행됩니다. 여러 구독자가 공유해야 하면 SharedFlow나 StateFlow를 사용하세요.
flow { } 빌더 안에서 다른 CoroutineContext로 emit하면 예외가 발생합니다. context를 바꾸려면 flowOn()을 사용하세요.
06코루틴 예외 처리 — SupervisorJob
일반 Job은 자식 코루틴의 예외가 부모와 형제를 모두 취소하지만, SupervisorJob은 실패한 자식만 취소합니다. 독립적인 작업들을 병렬로 실행할 때 한 작업의 실패가 전체에 영향을 주지 않도록 설계할 수 있습니다.
Kotlin code
import kotlinx.coroutines.*
fun main() = runBlocking {
println("=== 일반 Job: 하나 실패 → 전체 취소 ===")
try {
coroutineScope {
launch {
delay(100)
throw RuntimeException("작업1 실패!")
}
launch {
delay(200)
println("작업2 완료") // 실행되지 않음
}
}
} catch (e: RuntimeException) {
println("예외 포착: ${e.message}")
}
println()
println("=== SupervisorJob: 실패한 자식만 취소 ===")
supervisorScope {
val job1 = launch {
delay(100)
throw RuntimeException("작업A 실패!")
}
val job2 = launch {
delay(200)
println("작업B 완료!") // 정상 실행됨
}
// 개별 예외 처리
job1.join()
job2.join()
}
println()
println("=== CoroutineExceptionHandler 활용 ===")
val handler = CoroutineExceptionHandler { _, exception ->
println("전역 핸들러: ${exception.message}")
}
val supervisor = SupervisorJob()
val scope = CoroutineScope(coroutineContext + supervisor + handler)
scope.launch { throw ArithmeticException("계산 오류") }
scope.launch {
delay(100)
println("다른 작업 정상 실행!")
}
delay(200)
supervisor.cancel()
}supervisorScope는 coroutineScope의 supervisor 버전입니다. 자식 코루틴의 예외를 개별적으로 try-catch로 처리할 수 있으며, CoroutineExceptionHandler를 함께 사용하면 전역 에러 로깅이 가능합니다.
SupervisorJob은 직접 자식에게만 적용됩니다. 손자 코루틴은 일반 Job 규칙을 따르므로, 중첩된 구조에서는 각 레벨에 supervisor 전략을 명시적으로 적용해야 합니다.
07Flow 연산자 (Flow Operators)
Flow의 중간 연산자를 활용하여 비동기 데이터 스트림을 변환합니다. map, filter, transform, debounce 등을 다룹니다.
Kotlin code
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun sensorReadings(): Flow<Int> = flow {
val values = listOf(20, 25, 22, 35, 18, 40, 22)
for (v in values) {
delay(100)
emit(v)
}
}
fun main() = runBlocking {
// map + filter
sensorReadings()
.filter { it > 20 }
.map { "온도: ${it}°C" }
.collect { println(it) }
println("---")
// transform: 여러 값 방출 가능
sensorReadings()
.transform { value ->
emit(value)
if (value > 30) emit(-1) // 경고 신호
}
.collect { println(it) }
println("---")
// take, drop
sensorReadings()
.drop(2)
.take(3)
.collect { print("$it ") }
}transform은 map이나 filter를 일반화한 것입니다. 원소당 0개, 1개, 또는 여러 개를 방출할 수 있습니다.
Flow 연산자 내에서 delay를 호출할 수 있지만, 무한 Flow에 take 없이 collect하면 영원히 멈추지 않습니다.
08StateFlow와 SharedFlow
StateFlow는 상태를 보유하고, SharedFlow는 이벤트를 브로드캐스트합니다. 핫 스트림의 두 가지 핵심 타입입니다.
Kotlin code
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
class CounterViewModel {
private val _count = MutableStateFlow(0)
val count: StateFlow<Int> = _count.asStateFlow()
private val _events = MutableSharedFlow<String>()
val events: SharedFlow<String> = _events.asSharedFlow()
fun increment() {
_count.value++
}
suspend fun notifyChange() {
_events.emit("카운트 변경: ${_count.value}")
}
}
fun main() = runBlocking {
val vm = CounterViewModel()
// StateFlow 구독
val job1 = launch {
vm.count.collect { println("상태: $it") }
}
// SharedFlow 구독
val job2 = launch {
vm.events.collect { println("이벤트: $it") }
}
delay(100)
vm.increment()
vm.notifyChange()
vm.increment()
vm.notifyChange()
delay(100)
job1.cancel()
job2.cancel()
}StateFlow는 항상 최신 값을 유지하고, SharedFlow는 구독자가 없으면 값을 버립니다. 상태에는 StateFlow, 이벤트에는 SharedFlow를 사용하세요.
StateFlow는 동일 값을 연속으로 방출하지 않습니다(distinctUntilChanged). 같은 값이라도 알림이 필요하면 SharedFlow를 사용하세요.
09채널 (Channel)
채널은 코루틴 간 통신을 위한 동시성 기본 요소입니다. 생산자-소비자 패턴을 안전하게 구현합니다.
Kotlin code
import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
fun main() = runBlocking {
// 기본 채널
val channel = Channel<Int>(capacity = 5)
// 생산자
launch {
for (i in 1..5) {
println("전송: $i")
channel.send(i)
delay(100)
}
channel.close()
}
// 소비자
for (value in channel) {
println("수신: $value")
}
// produce 빌더
val numbers = produce {
var n = 1
while (true) {
send(n++)
delay(200)
}
}
// 5개만 수신
repeat(5) {
println("produce: ${numbers.receive()}")
}
numbers.cancel()
println("완료")
}Channel.CONFLATED는 최신 값만 유지하고, Channel.UNLIMITED는 무제한 버퍼링합니다. 용도에 맞는 버퍼 전략을 선택하세요.
채널을 close()하지 않으면 수신자의 for 루프가 영원히 대기합니다. 생산이 끝나면 반드시 닫으세요.
10구조화된 동시성 (Structured Concurrency)
코루틴의 생명주기를 구조적으로 관리합니다. 부모-자식 관계로 코루틴이 누수되는 것을 방지합니다.
Kotlin code
import kotlinx.coroutines.*
suspend fun fetchUserData(userId: String): String {
delay(300)
return "사용자 데이터: $userId"
}
suspend fun fetchOrders(userId: String): List<String> {
delay(200)
return listOf("주문1", "주문2")
}
suspend fun loadDashboard(userId: String) = coroutineScope {
// 병렬 실행 - 하나가 실패하면 모두 취소
val userData = async { fetchUserData(userId) }
val orders = async { fetchOrders(userId) }
println(userData.await())
println("주문: ${orders.await()}")
}
fun main() = runBlocking {
// 타임아웃 포함
try {
withTimeout(1000) {
loadDashboard("user-001")
}
} catch (e: TimeoutCancellationException) {
println("타임아웃!")
}
// withContext로 디스패처 전환
val result = withContext(Dispatchers.Default) {
(1..1_000_000).sum()
}
println("합계: $result")
}coroutineScope는 모든 자식이 완료될 때까지 대기하고, 하나가 실패하면 나머지를 취소합니다. 이것이 구조화된 동시성의 핵심입니다.
GlobalScope.launch는 구조화된 동시성을 깨뜨립니다. 항상 coroutineScope나 특정 스코프 내에서 코루틴을 시작하세요.
11코루틴 스코프 (CoroutineScope)
컴포넌트의 생명주기에 맞는 코루틴 스코프를 생성하여 리소스 누수를 방지합니다.
Kotlin code
import kotlinx.coroutines.*
class NetworkService : CoroutineScope {
private val job = SupervisorJob()
override val coroutineContext = Dispatchers.IO + job
fun fetchData(url: String) = launch {
println("[${Thread.currentThread().name}] 요청: $url")
delay(500)
println("응답 완료: $url")
}
fun destroy() {
println("서비스 종료 중...")
job.cancel()
}
}
class ViewModelScope {
private val scope = CoroutineScope(
SupervisorJob() + Dispatchers.Main.immediate
)
fun loadData() {
scope.launch {
try {
val data = withContext(Dispatchers.IO) {
delay(300)
"로드된 데이터"
}
println("UI 업데이트: $data")
} catch (e: CancellationException) {
println("취소됨")
}
}
}
fun clear() = scope.cancel()
}
fun main() = runBlocking {
val service = NetworkService()
service.fetchData("/api/users")
service.fetchData("/api/orders")
delay(600)
service.destroy()
}SupervisorJob을 사용하면 자식 코루틴의 실패가 다른 자식에게 전파되지 않아 독립적인 작업에 적합합니다.
스코프를 취소하지 않으면 코루틴이 계속 실행되어 메모리 누수가 발생합니다. onDestroy나 onCleared에서 반드시 취소하세요.
12취소와 타임아웃 (Cancellation & Timeout)
코루틴의 협력적 취소 메커니즘과 타임아웃 처리를 다룹니다. 안전한 리소스 정리 패턴을 포함합니다.
Kotlin code
import kotlinx.coroutines.*
suspend fun longRunningTask(): String {
try {
repeat(10) { i ->
println("작업 중: ${i + 1}/10")
delay(200) // 취소 지점
}
return "완료"
} finally {
// 취소되어도 실행됨
withContext(NonCancellable) {
println("리소스 정리 중...")
delay(100)
println("정리 완료")
}
}
}
fun main() = runBlocking {
// 수동 취소
val job = launch { longRunningTask() }
delay(500)
println("취소 요청")
job.cancelAndJoin()
println("---")
// 타임아웃 (null 반환)
val result = withTimeoutOrNull(500) {
longRunningTask()
}
println("결과: ${result ?: "타임아웃"}")
println("---")
// isActive 확인 (CPU 바운드 작업)
val cpuJob = launch(Dispatchers.Default) {
var i = 0
while (isActive) {
i++
if (i % 1000000 == 0) println("반복: $i")
}
println("CPU 작업 종료")
}
delay(100)
cpuJob.cancelAndJoin()
}withTimeoutOrNull은 타임아웃 시 예외 대신 null을 반환하여 더 안전합니다.
CPU 바운드 작업은 delay 같은 중단점이 없으면 취소에 응답하지 않습니다. isActive를 직접 확인하거나 ensureActive()를 호출하세요.
13뮤텍스와 세마포어 (Mutex & Semaphore)
코루틴 환경에서 공유 자원에 대한 동기화를 처리합니다. Mutex는 상호 배제, Semaphore는 동시 접근 수를 제한합니다.
Kotlin code
import kotlinx.coroutines.*
import kotlinx.coroutines.sync.*
fun main() = runBlocking {
// Mutex: 하나의 코루틴만 접근
val mutex = Mutex()
var counter = 0
val jobs = List(100) {
launch(Dispatchers.Default) {
repeat(1000) {
mutex.withLock {
counter++
}
}
}
}
jobs.forEach { it.join() }
println("카운터: $counter") // 정확히 100000
// Semaphore: 동시 접근 수 제한
val semaphore = Semaphore(3) // 최대 3개 동시
val apiJobs = List(10) { i ->
launch {
semaphore.withPermit {
println("API 호출 시작: $i")
delay(500)
println("API 호출 완료: $i")
}
}
}
apiJobs.forEach { it.join() }
}Mutex는 코루틴을 중단시키지만 스레드를 차단하지 않습니다. synchronized와 달리 코루틴에 안전합니다.
synchronized를 코루틴 내에서 사용하면 스레드가 차단되어 다른 코루틴이 실행되지 못합니다. 반드시 Mutex를 사용하세요.
14액터 모델 (Actor Model)
액터는 메시지 기반으로 상태를 캡슐화하는 동시성 패턴입니다. 채널을 통해 직렬화된 접근을 보장합니다.
Kotlin code
import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
sealed class CounterMsg {
data object Increment : CounterMsg()
data class Get(val response: CompletableDeferred<Int>) : CounterMsg()
}
fun CoroutineScope.counterActor() = actor<CounterMsg> {
var count = 0
for (msg in channel) {
when (msg) {
is CounterMsg.Increment -> count++
is CounterMsg.Get -> msg.response.complete(count)
}
}
}
fun main() = runBlocking {
val counter = counterActor()
// 동시에 증가
val jobs = List(100) {
launch {
repeat(1000) {
counter.send(CounterMsg.Increment)
}
}
}
jobs.forEach { it.join() }
// 결과 조회
val response = CompletableDeferred<Int>()
counter.send(CounterMsg.Get(response))
println("카운터: ${response.await()}") // 100000
counter.close()
}액터 패턴은 뮤텍스 없이도 스레드 안전합니다. 모든 상태 변경이 채널을 통해 직렬화되기 때문입니다.
actor 빌더는 실험적 API입니다. 프로덕션에서는 Mutex를 사용하는 것이 더 안정적일 수 있습니다.
15병렬 분해 (Parallel Decomposition)
async와 awaitAll로 작업을 병렬로 분해하고 결과를 합칩니다.
Kotlin code
import kotlinx.coroutines.*
data class ProductInfo(val name: String, val price: Int, val stock: Int)
suspend fun fetchName(id: Int): String {
delay(300); return "상품-$id"
}
suspend fun fetchPrice(id: Int): Int {
delay(200); return id * 1000
}
suspend fun fetchStock(id: Int): Int {
delay(250); return (10..100).random()
}
suspend fun getProductInfo(id: Int): ProductInfo = coroutineScope {
val name = async { fetchName(id) }
val price = async { fetchPrice(id) }
val stock = async { fetchStock(id) }
ProductInfo(name.await(), price.await(), stock.await())
}
suspend fun getMultipleProducts(ids: List<Int>): List<ProductInfo> =
coroutineScope {
ids.map { id -> async { getProductInfo(id) } }.awaitAll()
}
fun main() = runBlocking {
val start = System.currentTimeMillis()
val products = getMultipleProducts(listOf(1, 2, 3, 4, 5))
products.forEach { println(it) }
val elapsed = System.currentTimeMillis() - start
println("소요 시간: ${elapsed}ms") // ~300ms (병렬)
}awaitAll()은 모든 Deferred의 결과를 기다립니다. 하나가 실패하면 나머지도 취소됩니다.
async를 coroutineScope 밖에서 사용하면 구조화된 동시성이 깨집니다. 항상 스코프 내에서 사용하세요.
16코루틴 테스트 (Coroutine Testing)
runTest와 TestDispatcher로 코루틴 코드를 결정적으로 테스트합니다.
Kotlin code
import kotlinx.coroutines.*
import kotlinx.coroutines.test.*
class DataRepository {
suspend fun fetchData(): String {
delay(5000) // 실제 네트워크 지연
return "서버 데이터"
}
}
class DataViewModel(private val repo: DataRepository) {
var data: String = ""
private set
var isLoading: Boolean = false
private set
suspend fun loadData() {
isLoading = true
data = repo.fetchData()
isLoading = false
}
}
// 테스트 코드
fun main() = runTest {
val repo = DataRepository()
val viewModel = DataViewModel(repo)
// runTest는 delay를 자동으로 건너뜀
val job = launch { viewModel.loadData() }
// advanceUntilIdle: 모든 코루틴 완료까지 진행
advanceUntilIdle()
println("데이터: ${viewModel.data}")
println("로딩: ${viewModel.isLoading}")
// delay(5000)이 즉시 실행됨!
println("테스트 시간: ${currentTime}ms")
}runTest는 가상 시간을 사용하므로 delay(5000)이 실제로 5초를 기다리지 않습니다. 테스트가 즉시 완료됩니다.
runBlocking으로 코루틴을 테스트하면 실제 지연이 발생합니다. 코루틴 테스트에는 항상 runTest를 사용하세요.
17예외 전파 (Exception Propagation)
코루틴의 예외 전파 메커니즘을 이해합니다. launch와 async의 예외 처리 차이를 다룹니다.
Kotlin code
import kotlinx.coroutines.*
fun main() = runBlocking {
// CoroutineExceptionHandler
val handler = CoroutineExceptionHandler { _, e ->
println("핸들러: ${e.message}")
}
// launch: 예외가 즉시 전파
val scope = CoroutineScope(SupervisorJob() + handler)
scope.launch {
throw RuntimeException("launch 예외")
}
delay(100)
// async: await() 호출 시 예외 전파
val deferred = scope.async {
throw RuntimeException("async 예외")
}
try {
deferred.await()
} catch (e: Exception) {
println("await 캐치: ${e.message}")
}
// supervisorScope 내 개별 처리
supervisorScope {
val child1 = launch {
throw RuntimeException("자식1 실패")
}
val child2 = launch {
delay(100)
println("자식2 정상 실행")
}
child1.join()
child2.join()
}
}CoroutineExceptionHandler는 launch에서만 동작합니다. async의 예외는 await() 호출부에서 try-catch로 처리합니다.
coroutineScope 내에서 자식이 실패하면 모든 형제도 취소됩니다. 독립적인 실패 처리가 필요하면 supervisorScope를 사용하세요.
18callbackFlow로 콜백 래핑
callbackFlow로 콜백 기반 API를 Flow로 변환합니다. 리스너 등록/해제를 안전하게 관리합니다.
Kotlin code
import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
import kotlinx.coroutines.flow.*
// 콜백 기반 API 시뮬레이션
interface LocationListener {
fun onLocationChanged(lat: Double, lng: Double)
fun onError(message: String)
}
class LocationManager {
private var listener: LocationListener? = null
fun register(l: LocationListener) { listener = l; println("리스너 등록") }
fun unregister() { listener = null; println("리스너 해제") }
fun simulateUpdates() {
listener?.onLocationChanged(37.5665, 126.9780)
listener?.onLocationChanged(37.5670, 126.9785)
}
}
data class Location(val lat: Double, val lng: Double)
fun LocationManager.locationFlow(): Flow<Location> = callbackFlow {
val listener = object : LocationListener {
override fun onLocationChanged(lat: Double, lng: Double) {
trySend(Location(lat, lng))
}
override fun onError(message: String) {
close(Exception(message))
}
}
register(listener)
awaitClose { unregister() }
}
fun main() = runBlocking {
val manager = LocationManager()
val job = launch {
manager.locationFlow().collect { println("위치: $it") }
}
delay(100)
manager.simulateUpdates()
delay(100)
job.cancel()
}awaitClose 블록에서 리스너를 해제하면 Flow가 취소될 때 자동으로 정리됩니다. 리소스 누수를 방지하는 핵심 패턴입니다.
callbackFlow 내에서 awaitClose를 생략하면 채널이 즉시 닫힙니다. 반드시 awaitClose로 Flow의 수명을 관리하세요.
19SupervisorScope 심화
supervisorScope와 SupervisorJob의 차이점과 올바른 사용법을 다룹니다.
Kotlin code
import kotlinx.coroutines.*
fun main() = runBlocking {
println("=== supervisorScope ===")
supervisorScope {
val job1 = launch {
delay(100)
throw RuntimeException("작업1 실패")
}
val job2 = launch {
delay(200)
println("작업2 완료!") // 실행됨
}
val job3 = launch {
delay(300)
println("작업3 완료!") // 실행됨
}
// 개별 예외 처리
try { job1.join() } catch (e: Exception) {
println("job1 오류: ${e.message}")
}
job2.join()
job3.join()
}
println("
=== 개별 try-catch ===")
coroutineScope {
val results = listOf("A", "B", "ERROR", "C").map { item ->
async(SupervisorJob()) {
if (item == "ERROR") throw RuntimeException("$item 실패")
delay(100)
"처리: $item"
}
}
results.forEach { deferred ->
runCatching { deferred.await() }
.onSuccess { println(it) }
.onFailure { println("실패: ${it.message}") }
}
}
}supervisorScope는 자식의 실패를 격리합니다. 여러 독립 API 호출을 병렬 실행하고 일부 실패를 허용할 때 사용하세요.
async(SupervisorJob())으로 독립 Deferred를 만들면 부모와의 구조가 깨집니다. 가능하면 supervisorScope 내에서 async를 사용하세요.
20Flow 연산자 심화 (combine, merge)
여러 Flow를 결합하는 고급 연산자입니다. combine은 최신 값을, merge는 모든 방출을 합칩니다.
Kotlin code
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
val names = flow {
emit("김철수"); delay(300)
emit("이영희"); delay(300)
emit("박지성")
}
val scores = flow {
emit(85); delay(200)
emit(92); delay(200)
emit(78); delay(200)
emit(95)
}
// combine: 양쪽 최신 값 조합
println("=== combine ===")
names.combine(scores) { name, score ->
"$name: $score점"
}.collect { println(it) }
// zip: 1:1 매칭
println("=== zip ===")
names.zip(scores) { name, score ->
"$name: $score점"
}.collect { println(it) }
// merge: 모든 방출 합치기
println("=== merge ===")
val flow1 = flowOf("A", "B").onEach { delay(100) }
val flow2 = flowOf("1", "2", "3").onEach { delay(150) }
merge(flow1, flow2).collect { print("$it ") }
}combine은 어느 한쪽이 새 값을 방출할 때마다 최신 조합을 만듭니다. UI에서 여러 상태를 합칠 때 유용합니다.
zip은 양쪽 모두 방출해야 결합합니다. 한쪽이 완료되면 나머지도 무시되므로 길이가 다른 Flow에 주의하세요.
21Flow 수집 연산자 (Terminal Operators)
Flow의 터미널 연산자인 collect, toList, first, reduce, launchIn 등을 다룹니다.
Kotlin code
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun numbers(): Flow<Int> = flow {
for (i in 1..10) {
delay(50)
emit(i)
}
}
fun main() = runBlocking {
// toList: 전체 수집
val list = numbers().toList()
println("리스트: $list")
// first / firstOrNull
val first = numbers().first { it > 5 }
println("첫 번째 >5: $first")
// reduce / fold
val sum = numbers().reduce { acc, value -> acc + value }
println("합: $sum")
val product = numbers().take(5).fold(1) { acc, value -> acc * value }
println("곱(1~5): $product")
// count
val evenCount = numbers().count { it % 2 == 0 }
println("짝수 개수: $evenCount")
// launchIn: 별도 코루틴에서 수집
val job = numbers()
.onEach { print("$it ") }
.onCompletion { println("
수집 완료") }
.launchIn(this)
job.join()
}launchIn은 scope.launch { flow.collect {} }의 축약형입니다. UI 이벤트 처리에 자주 사용됩니다.
first()에 조건을 만족하는 원소가 없으면 NoSuchElementException이 발생합니다. 안전하게 firstOrNull()을 사용하세요.
정리하며
- 모든 코루틴은 스코프에 속하게 하고, GlobalScope는 수명이 프로세스와 같을 때만 씁니다
- 형제 작업을 독립적으로 굴리려면 Job 대신 SupervisorJob 또는 supervisorScope를 씁니다
- CPU 계산 루프에는 ensureActive()나 yield()를 넣어야 취소가 실제로 먹힙니다
- catch로 CancellationException을 삼키지 말고, 정리는 NonCancellable 안에서 합니다
더 깊이 들어가고 싶다면 Kotlin 학습 라이브러리에서 다른 주제 가이드를 이어서 보거나, 언어 비교에서 같은 개념이 다른 언어에서 어떻게 표현되는지 확인해 보세요.