JVM 동시성 모델 이해하기 (5) - Kotlin Coroutines

// Reactor — 깊은 flatMap 체이닝
public Mono<UserDashboard> buildDashboard(Long userId) {
    return userRepository.findById(userId)
        .flatMap(user -> orderRepository.findByUserId(user.getId())
            .flatMap(order -> productRepository.findById(order.getProductId())
                .flatMap(product -> reviewRepository.findByProductId(product.getId())
                    .collectList()
                    .map(reviews -> new ProductWithReviews(product, reviews))))
            .collectList()
            .map(products -> new OrderDetail(user, products)))
        .flatMap(detail -> pointRepository.findByUserId(userId)
            .map(points -> new UserDashboard(detail, points)));
}

// Coroutine — 같은 논블로킹, 동기 코드 형태
suspend fun buildDashboard(userId: Long): UserDashboard {
    val user = userRepository.findById(userId)
    val orders = orderRepository.findByUserId(user.id)
    val products = orders.map { order ->
        val product = productRepository.findById(order.productId)
        val reviews = reviewRepository.findByProductId(product.id)
        ProductWithReviews(product, reviews)
    }
    val points = pointRepository.findByUserId(userId)
    return UserDashboard(OrderDetail(user, products), points)
}

suspend fun fetchUser(id: Long): User {
    delay(1000)  // 1초 대기 — 스레드를 블로킹하지 않음
    return User(id, "Alice")
}

// 컴파일 에러
fun main() {
    fetchUser(1L)  // Error: Suspend function should be called only from a coroutine
}

// OK
fun main() = runBlocking {
    val user = fetchUser(1L)
    println(user)
}

val scope = CoroutineScope(Dispatchers.Default)

scope.launch {
    val user = fetchUser(1L)
    saveToCache(user)
    println("캐시 저장 완료")
}
// launch는 즉시 반환 — 코루틴은 백그라운드에서 실행

val job = scope.launch {
    delay(2000)
    println("작업 완료")
}

job.join()  // 코루틴이 끝날 때까지 기다림
println("job 완료 후 실행됨")

// 또는 취소
val longJob = scope.launch {
    repeat(1000) { i ->
        println("작업 $i")
        delay(500)
    }
}
delay(2000)
longJob.cancel()

val deferred: Deferred<User> = scope.async {
    fetchUser(1L)  // User를 반환
}

val user: User = deferred.await()  // 결과가 준비될 때까지 중단

suspend fun getUserWithOrders(userId: Long): UserWithOrders = coroutineScope {
    // 두 작업을 동시에 시작
    val userDeferred = async { fetchUser(userId) }
    val ordersDeferred = async { fetchOrders(userId) }

    // 둘 다 완료되면 결합
    UserWithOrders(userDeferred.await(), ordersDeferred.await())
}

fun main() = runBlocking {
    // 여기서부터 코루틴 세계
    val user = fetchUser(1L)
    println(user)
}

suspend fun processAll() = coroutineScope {
    launch { task1() }
    launch { task2() }
    launch { task3() }
    // 세 작업이 모두 완료될 때까지 여기서 중단
}
// processAll() 이후 — 세 작업 모두 완료됨이 보장

// runBlocking — 스레드 자체를 잡고 안 놓아줌
fun main() = runBlocking {  // main 스레드가 여기서 멈춤
    launch { delay(1000) }
    // main 스레드는 이 블록이 끝날 때까지 다른 일을 못 함
}

// coroutineScope — 코루틴만 중단, 스레드는 반환
suspend fun process() = coroutineScope {  // 코루틴이 중단됨
    launch { delay(1000) }
    // 스레드는 반환되어 다른 코루틴을 실행할 수 있음
}

val scope = CoroutineScope(Dispatchers.Default)

scope.launch {           // 부모 코루틴
    launch { task1() }   // 자식 코루틴 1
    launch { task2() }   // 자식 코루틴 2
}

// Reactor
fun readFile(): Mono<String> =
    Mono.fromCallable { File("data.txt").readText() }
        .subscribeOn(Schedulers.boundedElastic())

// Coroutine — 같은 의미
suspend fun readFile(): String = withContext(Dispatchers.IO) {
    File("data.txt").readText()  // 블로킹 I/O를 IO 스레드에서 실행
}

suspend fun process() {
    // Default 스레드에서 실행 중
    val data = withContext(Dispatchers.IO) {
        // IO 스레드로 전환
        readFromDatabase()
    }
    // 다시 Default 스레드로 복귀
    transform(data)
}

suspend fun fetchUserWithOrders(userId: Long): UserWithOrders {
    println("시작")                           // 중단점 없음
    val user = fetchUser(userId)              // 중단점 1
    println("유저 조회 완료: ${user.name}")
    val orders = fetchOrders(user.id)         // 중단점 2
    println("주문 조회 완료: ${orders.size}건")
    return UserWithOrders(user, orders)
}

// 우리가 작성한 코드
suspend fun fetchUser(userId: Long): User

// 컴파일러가 변환한 코드 (개념적 예시)
fun fetchUser(userId: Long, continuation: Continuation<User>): Any?

interface Continuation<in T> {
    val context: CoroutineContext
    fun resumeWith(result: Result<T>)
}

// fetchUser의 변환된 내부 (개념적 의사코드)
fun fetchUser(userId: Long, cont: Continuation<User>): Any? {
    // 네트워크 요청 시작
    val pending = networkClient.requestAsync("/users/$userId")

    if (pending.isCompleted) {
        // 이미 완료된 경우 (캐시 히트 등) → 결과를 직접 반환
        return pending.result  // User 객체
    } else {
        // 아직 응답 안 옴 → continuation을 콜백으로 등록하고 SUSPENDED 반환
        pending.onComplete { user ->
            cont.resumeWith(Result.success(user))  // 나중에 호출됨
        }
        return COROUTINE_SUSPENDED  // "지금은 결과 없음" 신호
    }
}

// 컴파일러가 생성한 상태 머신 (의사코드)
fun fetchUserWithOrders(userId: Long, cont: Continuation<*>): Any? {
    // 상태를 저장하는 객체 (최초 호출 시 생성)
    val sm = cont as? FetchUserWithOrdersSM ?: FetchUserWithOrdersSM(cont)

    when (sm.label) {
        0 -> {
            // 상태 0: 시작 ~ 첫 번째 중단점
            println("시작")
            sm.label = 1              // 다음 상태 설정
            sm.userId = userId        // 지역 변수 저장
            val result = fetchUser(userId, sm)  // sm을 콜백으로 전달
            if (result == COROUTINE_SUSPENDED) return COROUTINE_SUSPENDED
            sm.result = Result.success(result)
        }
        1 -> {
            // 상태 1: fetchUser 완료 후 ~ 두 번째 중단점
            val user = sm.result!!.getOrThrow() as User
            sm.user = user
            println("유저 조회 완료: ${user.name}")
            sm.label = 2
            val result = fetchOrders(user.id, sm)
            if (result == COROUTINE_SUSPENDED) return COROUTINE_SUSPENDED
            sm.result = Result.success(result)
        }
        2 -> {
            // 상태 2: fetchOrders 완료 후 ~ 함수 끝
            val orders = sm.result!!.getOrThrow() as List<Order>
            val user = sm.user
            println("주문 조회 완료: ${orders.size}건")
            return UserWithOrders(user, orders)
        }
    }
}

fun numbers(): Flow<Int> = flow {
    for (i in 1..5) {
        delay(100)       // 논블로킹 대기
        emit(i)          // 값 방출
    }
}

// 사용
suspend fun main() {
    numbers().collect { value ->  // collect = subscribe
        println(value)
    }
}

numbers()
    .filter { it % 2 == 0 }          // 짝수만
    .map { it * 10 }                  // 10배
    .collect { println(it) }          // 20, 40

flow {
    // IO 스레드에서 실행
    emit(readFromDatabase())
}
    .flowOn(Dispatchers.IO)        // 위쪽 flow의 실행 스레드를 지정
    .map { transform(it) }         // Default 스레드에서 실행
    .collect { println(it) }

// Coroutine — Backpressure 전략 없이 suspend가 자동 해결
flow {
    for (i in 1..1000) {
        emit(i)          // collect가 처리 중이면 여기서 자동 중단
    }
}.collect { value ->
    delay(1000)          // 느린 소비자
    println(value)
}

// SharedFlow — 이벤트용. 초기값 없음.
val events = MutableSharedFlow<ClickEvent>()

// StateFlow — 상태용. 초기값 필수.
val uiState = MutableStateFlow(UiState.Loading)

MutableSharedFlow<Int>(
    replay = 2,              // 새 collector에게 최근 2개 값을 재생
    extraBufferCapacity = 3  // emit이 suspend되지 않는 추가 버퍼
)

// Fan-out 예시: 여러 워커가 작업을 나눠 처리
val tasks = Channel<Task>(capacity = 100)  // 작업 큐

// 생산자 — 작업을 Channel에 넣음
launch {
    for (task in fetchPendingTasks()) {
        tasks.send(task)
    }
    tasks.close()
}

// 워커 3개가 작업을 나눠 받아 처리 (fan-out)
repeat(3) { workerId ->
    launch {
        for (task in tasks) {  // 각 task는 하나의 워커만 받음
            println("워커 $workerId 처리: ${task.id}")
            process(task)
        }
    }
}

val channel = Channel<Int>()

// 생산자 코루틴
launch {
    for (i in 1..5) {
        channel.send(i)       // 소비자가 받을 준비가 되면 전송
        println("보냄: $i")
    }
    channel.close()
}

// 소비자 코루틴
launch {
    for (value in channel) {  // 채널이 닫힐 때까지 반복
        println("받음: $value")
        delay(1000)           // 느린 소비자
    }
}

// 버퍼 없음 — send와 receive가 만날 때까지 둘 다 중단
val rendezvous = Channel<Int>(Channel.RENDEZVOUS)

// 고정 버퍼 — 버퍼가 차면 send 중단
val buffered = Channel<Int>(capacity = 10)

// 무제한 버퍼 — send가 중단되지 않음 (메모리 주의)
val unlimited = Channel<Int>(Channel.UNLIMITED)

// 최신 값만 유지 — 버퍼가 차면 오래된 값을 덮어씀
val conflated = Channel<Int>(Channel.CONFLATED)

suspend fun getUser(id: Long): User {
    return try {
        fetchUser(id)
    } catch (e: NetworkException) {
        getCachedUser(id)  // 폴백
    }
}

val handler = CoroutineExceptionHandler { _, exception ->
    println("처리되지 않은 예외: ${exception.message}")
}

val scope = CoroutineScope(Dispatchers.Default + handler)

scope.launch {
    throw RuntimeException("문제 발생!")
    // → handler가 잡아서 처리
}

suspend fun loadDashboard() = supervisorScope {
    val profile = async { fetchProfile() }       // 실패해도
    val notifications = async { fetchNotifications() }  // 이건 계속 실행
    val recommendations = async { fetchRecommendations() }

    DashboardData(
        profile = try { profile.await() } catch (e: Exception) { null },
        notifications = notifications.await(),
        recommendations = recommendations.await()
    )
}

val job = launch {
    try {
        repeat(1000) { i ->
            println("작업 $i")
            delay(500)  // 취소 가능한 중단점
        }
    } catch (e: CancellationException) {
        println("취소됨 — 정리 작업 수행")
        throw e  // 반드시 다시 던져야 취소가 전파됨
    }
}

delay(2000)
job.cancel()  // CancellationException 발생

// 방법 1: yield()로 취소 확인
suspend fun heavyComputation() = coroutineScope {
    var result = 0
    for (i in 1..1_000_000) {
        result += complexCalc(i)
        if (i % 1000 == 0) yield()  // 1000번마다 취소 확인
    }
    result
}

// 방법 2: isActive로 직접 확인
suspend fun heavyComputation2() = coroutineScope {
    var result = 0
    for (i in 1..1_000_000) {
        if (!isActive) break    // 취소되었으면 루프 탈출
        result += complexCalc(i)
    }
    result
}