fxts pipe로 이해하는 Kotlin Flow: 같아 보이지만 다른 지점들

// fxts pipe( userIds, map(id => userMap.get(id)), zipWithIndex, map(([index, user]) => ({ ...user, rank: index })), toArray, );

// Kotlin Flow userIds.asFlow() .map { id -> userMap[id] } .withIndex() .map { (index, user) -> user.copy(rank = index) } .toList()

fxts: pipe(source, map(), zipWithIndex, map(), toArray) Flow: source.asFlow(). map(). withIndex(). map(). toList()

// fxts await pipe( notifications, toAsync, concurrent(10), each(noti => sendPush(noti)), );

// Kotlin Flow notifications.asFlow() .flatMapMerge(10) { noti -> flow { emit(sendPush(noti)) } } .collect()

+---------------------------+ | Builder (생성) | | | | flowOf(1,2,3) | | .asFlow() | | flow { emit() } | +---------------------------+ | v +---------------------------+ | Operator (변환) | | | | .map { } | | .filter { } | | .flatMapMerge { } | +---------------------------+ | v +---------------------------+ | Terminal (소비) | | | | .collect { } | | .toList() | | .first() | +---------------------------+

// fxts: lazy iterable이지만, 한 번 소비하면 끝 const lazy = pipe([1, 2, 3], map(x => { console.log('변환:', x); return x * 2; })); [...lazy]; // "변환: 1", "변환: 2", "변환: 3" 출력 [...lazy]; // 아무것도 출력되지 않음. generator는 이미 소비됨.

// Flow: collect할 때마다 처음부터 다시 실행됨 val myFlow = flowOf(1, 2, 3) .map { println("변환: $it"); it * 2 } myFlow.toList() // "변환: 1", "변환: 2", "변환: 3" 출력 myFlow.toList() // "변환: 1", "변환: 2", "변환: 3" 또 출력. 처음부터 다시 실행.

입력: A -> B -> C 처리: [A 완료] -> [B 완료] -> [C 완료] 결과: a1, a2, b1, b2, c1, c2 (순서 보장)

입력: A, B, C, D, E (concurrency = 3) 시점1: [A 처리중] [B 처리중] [C 처리중] [D 대기] [E 대기] 시점2: [A 완료] [B 처리중] [C 처리중] [D 시작] [E 대기] 시점3: [B 완료] [C 완료] [D 처리중] [E 시작] 결과: 순서 보장 안 됨 (완료 순서대로 emit)

// fxts await pipe(items, toAsync, map(fn), concurrent(3), toArray);

// Kotlin Flow items.asFlow() .flatMapMerge(3) { item -> flow { emit(fn(item)) } } .toList()

입력: A -> B -> C 처리: [A 처리--취소!] [B 처리--취소!] [C 처리----완료] 결과: C의 결과만

Node.js (단일 스레드): 스레드1: [A의 count++] ... [B의 count++] ... [C의 count++] 이벤트루프가 번갈아 실행. 절대 겹치지 않음. JVM (멀티 스레드): 스레드1: [A의 count++] 스레드2: [B의 count++] <- 동시 접근! 값 유실 가능 스레드3: [C의 count++]

+--코루틴1: apiCall(A) -> emit(resultA)--+ orders asFlow -> +--코루틴2: apiCall(B) -> emit(resultB)--+-> Channel -> toList +--코루틴3: apiCall(C) -> emit(resultC)--+

// 에러! IllegalStateException flow { withContext(Dispatchers.IO) { emit(value) // 다른 context에서 emit 불가 } } // 올바른 방법: flowOn으로 외부에서 지정 flow { emit(value) } .flowOn(Dispatchers.IO)

flow { emit(heavyCpuWork()) } // (1) Default 풀에서 실행 .flowOn(Dispatchers.Default) // <- 여기가 경계선 .map { saveToDb(it) } // (2) IO 풀에서 실행 .flowOn(Dispatchers.IO) // <- 여기가 경계선 .collect { updateUi(it) } // (3) 호출자 스레드에서 실행

+-------------------------------+ | flow { heavyCpuWork() } | Dispatchers.Default (CPU 코어 수만큼) +-------------------------------+ flowOn(Default) <-- 경계선. 내부적으로 Channel로 안전하게 전환. +-------------------------------+ | .map { saveToDb(it) } | Dispatchers.IO (기본 최대 64개 스레드) +-------------------------------+ flowOn(IO) <-- 경계선 +-------------------------------+ | .collect { updateUi(it) } | 호출자의 스레드 +-------------------------------+

flow { emit(1) } flowOn(IO) collect { } | | | IO 스레드에서 Channel 호출자 스레드에서 값 생산 (thread-safe 큐) 값 소비 | | | send(1) -------------> [1] ---------> receive(1)

// fxts: 에러 처리가 map 안에서 수동 pipe( items, toAsync, map(async (item) => { try { return await pRetry(() => riskyApiCall(item), { retries: 3 }); } catch { return fallbackData; } }), toArray, );

// Flow: retry, catch가 파이프의 일부 flow { emit(riskyApiCall()) } .retry(3) { it is IOException } // 3번 재시도 .catch { emit(fallbackData) } // 실패시 대체 값 .collect { save(it) }

fun fetchAllPages(): Flow<Item> = flow { var page = 0 var hasMore = true while (hasMore) { val response = api.getItems(page = page, size = 100) response.items.forEach { emit(it) } hasMore = response.hasNext page++ } } // 50개만 필요하면 나머지 페이지의 API를 호출하지 않는다 fetchAllPages() .filter { it.isActive } .take(50) .toList()

orders.asFlow() .flatMapMerge(5) { order -> flow { emit(externalApi.getItems(order.id)) } } .catch { emit(emptyList()) } // 하나라도 실패하면 여기서 중단 .toList()

orders.asFlow() .flatMapMerge(5) { order -> flow { val items = runCatching { externalApi.getItems(order.id) }.getOrElse { emptyList() } emit(order.copy(items = items)) } } .toList()

// fxts: 항목별 에러 격리 const results = await pipe( orders, toAsync, map(async (order) => { try { const items = await externalApi.getItems(order.id); return { ...order, items }; } catch { return { ...order, items: [] }; } }), concurrent(5), toArray, );

suspend -> 논블로킹 (스레드 반환). Node.js의 await와 동일. Flow -> suspend 위에서 선언적 파이프 스타일 코드 작성. fxts의 pipe와 동일. flatMapMerge -> 파이프 안에서 동시성 제어. fxts의 concurrent(n)와 동일.

@GetMapping("/delivery") suspend fun getDelivery(): DeliveryInfo { return webClient.get().retrieve().awaitBody() // ^^^^ 여기서 suspend -> 스레드 반환 -> 응답 오면 resume }

스레드의 움직임을 보면 이렇다. 스레드1: [요청 받음] [외부 API 호출 -> suspend -> 스레드1 반환!] 스레드1: [다른 요청 처리] <- 반환된 스레드가 즉시 다른 일을 함 스레드2: [외부 API 응답 도착 -> resume -> 반환]

suspend 없이 (블로킹): 스레드1: [요청1 - 외부API 대기중.........응답] 스레드2: [요청2 - 외부API 대기중.........응답] ... 스레드200: [요청200 - 외부API 대기중.......응답] 요청201: 스레드 없음! 대기! <- 병목 suspend 있을 때 스레드1: [요청1 시작 -> suspend] [요청53 시작 -> suspend] [요청99 시작 -> suspend] 스레드2: [요청2 시작 -> suspend] [요청54 시작 -> suspend] [요청100 시작 -> suspend] ... 200개 스레드가 돌아가면서 수천 개 요청을 처리 가능

// 방법 1: for + suspend (명령형) suspend fun updateDeliveries(orderIds: List<String>) { for (orderId in orderIds) { val delivery = externalApi.get(orderId) // suspend -> 스레드 반환 db.update(delivery) // suspend -> 스레드 반환 } } // 방법 2: Flow (선언적 파이프) suspend fun updateDeliveries(orderIds: List<String>) { orderIds.asFlow() .map { externalApi.get(it) } // suspend -> 스레드 반환 .map { db.update(it) } // suspend -> 스레드 반환 .collect() }

방법 1 (for + suspend): 스레드1: [API호출 -> suspend] [반환] ... [resume -> DB저장 -> suspend] [반환] 방법 2 (Flow): 스레드1: [API호출 -> suspend] [반환] ... [resume -> DB저장 -> suspend] [반환] 동일함!

// for문: 순차 처리만 가능 for (orderId in orderIds) { val delivery = externalApi.get(orderId) // 하나 끝나야 다음 db.update(delivery) } // Flow + flatMapMerge: 동시성 제어 병렬 orderIds.asFlow() .flatMapMerge(10) { orderId -> flow { val delivery = externalApi.get(orderId) emit(db.update(delivery)) } } .collect()

// fxts: concurrent 없으면 for문과 성능 같음 pipe(items, toAsync, map(fn), toArray); // fxts: concurrent 붙으면 병렬 처리 pipe(items, toAsync, map(fn), concurrent(10), toArray);

Flow 순차 (map -> collect) = for + suspend -> 같은 성능. 가독성 차이만. Flow 병렬 (flatMapMerge -> collect) = concurrent(n) -> 이게 Flow의 진짜 성능 이점.

suspend fun getDelivery(orderId: String): DeliveryInfo { return externalApi.get(orderId) // suspend 지점 -> 스레드 반환 }

orderIds.asFlow() .flatMapMerge(3) { id -> flow { emit(externalApi.get(id)) } } .filter { it.hasChanged } .toList()

// Flow로 억지로 하면... 어색함 val results = flowOf("order", "user", "products") .flatMapMerge(3) { type -> flow { when (type) { "order" -> emit(fetchOrder(orderId)) // Order 타입 "user" -> emit(fetchUser(userId)) // User 타입 "products" -> emit(fetchProducts(ids)) // List<Product> 타입 } } } .toList() // results[0]이 order인지 user인지 보장 안 됨! // 타입도 전부 Any로 퉁쳐야 함

// coroutineScope + async: 자연스러움 coroutineScope { val order = async { fetchOrder(orderId) } // Order 타입 val user = async { fetchUser(userId) } // User 타입 val products = async { fetchProducts(ids) } // List<Product> 타입 OrderDetail( order = order.await(), // 타입 안전 user = user.await(), // 타입 안전 products = products.await() // 타입 안전 ) }

단건 비동기 호출 -> suspend fun 같은 작업 N건 병렬 + 파이프 변환 -> Flow + flatMapMerge 다른 작업 동시 실행 후 합치기 -> coroutineScope + async 실시간 스트리밍 (SSE 등) -> Flow 필수

[HTTP 서버 레이어] -> 요청을 어떻게 받는가? (Tomcat vs Netty) [비즈니스 로직 레이어] -> 받은 요청을 어떻게 처리하는가? (블로킹 vs suspend/Flow)

Spring MVC (전통): 웨이터 200명. 손님 1명당 웨이터 1명 배정. 웨이터가 주방에 주문 넣고 음식 나올 때까지 테이블 옆에서 서서 대기. 손님 201명째 -> 웨이터 없어서 입구에서 대기. Spring MVC + suspend: 웨이터 200명. 주문 넣고 다른 테이블로 이동. 음식 나오면 아무 웨이터나 가져다줌. 200명 웨이터로 수천 명 처리 가능. Spring WebFlux: 웨이터 4명. 주문 메모만 하고 바로 다음 손님. 주방에서 음식 나오면 아무 웨이터나 가져다줌. 손님 10000명이어도 처리 가능. Node.js Express와 유사한 모델.