[코틀린 코루틴] 3장. 중단은 어떻게 작동할까?

중단 함수(suspend function)는 코틀린 코루틴의 핵심이며, 코틀린 코루틴의 다른 모든 개념의 기초가 되는 요소입니다.

코루틴은 중단되었을 때 Contiunation 객체를 반환합니다.

Continuation을 이용하면 멈췄던 곳에서 다시 코루틴을 실행할 수 있습니다.

스레드는 저장이 불가능하고 멈추는 것만 가능하지만, 코루틴은 중단했을 때 어떠한 자원도 사용하지 않습니다.

코루틴은 다른 스레드에서 시작할 수 있고, Continuation 객체는 (이론상) 직렬화와 역직렬화가 가능하며 다시 실행될 수 있습니다.

 

 

재개

 

중단 가능한 main 함수를 통해 작업이 재개되는 원리를 살펴보겠습니다.

suspend fun main() {
    println("Before")

    println("After")
}
// Before
// After

Before와 After 사이를 중단하기 위해 코루틴 중단 함수인 suspendCoroutine 함수를 사용합니다.

suspend fun main() {
    println("Before")
    suspendCoroutine<Unit> { }
    println("After")
}
// Before

이렇게 하면, After는 출력되지 않으며 메인 함수가 끝나지 않았기 때문에 코드는 실행된 상태로 유지됩니다.

suspendCoroutine 함수의 인자로 들어가는 람다 함수가 있는데, 프로그램을 재개하기 위해서는 이 람다 함수가 인자로 받고있는 Continuation 객체의 resume 함수를 사용하면 재개할 수 있습니다.

suspend fun main() {
    println("Before")
    suspendCoroutine<Unit> { continuation ->
        continuation.resume(Unit)
    }
    println("After")
}
// Before
// After

resume함수는 코틀린에서 제공하는 Continuation의 확장 API입니다.

내부적으로는 resumeWith을 사용하고 있으며, resumeWith은 인자로 Result타입을 받습니다.

public interface Continuation<in T> {

    public val context: CoroutineContext

    public fun resumeWith(result: Result<T>)
}

public inline fun <T> Continuation<T>.resume(value: T): Unit =
    resumeWith(Result.success(value))

정해진 시간 뒤에 코루틴을 다시 재개하는 함수를 만들 수 있으며, 이때 Continuation 객체는 다음 코드와 같이 람다 표현식이 통제합니다.

fun continuationAfterSecond(continuation: Continuation<Unit>) {
    thread {
        Thread.sleep(1000)
        continuation.resume(Unit)
    }
}

suspend fun main() {
    println("Before")

    suspendCoroutine<Unit> { continuation ->
        continuationAfterSecond(continuation)
    }

    println("After")
}
// Before
// (1초 후)
// After

스레드를 생성하는 비용이 상당히 많이 드는데, 만들어진 다음 1초 뒤에 사라지는 스레드는 불필요해 보입니다.

정해진 시간이 지나면 continuation.resume(Unit)을 호출하도록 알람을 설정하는 ScheduledExecutorService를 사용할 수도 있습니다.

private val executor =
    Executors.newSingleThreadScheduledExecutor {
        Thread(it, "scheduler").apply { isDaemon = true }
    }

suspend fun main() {
    println("Before")

    suspendCoroutine<Unit> { continuation ->
        executor.schedule({
            continuation.resume(Unit)
        }, 1000, TimeUnit.MILLISECONDS)
    }

    println("After")
}
// Before
// (1초 후)
// After

중단할 수 있는 기능을 delay 함수로 추출해보면 다음과 같습니다.

private val executor =
    Executors.newSingleThreadScheduledExecutor() {
        Thread(it, "scheduler").apply { isDaemon = true }
    }

suspend fun delay(timeMillis: Long): Unit =
    suspendCoroutine { cont ->
        executor.schedule({
            cont.resume(Unit)
        }, timeMillis, TimeUnit.MILLISECONDS)
    }

suspend fun main() {
    println("Before")

    delay(1000)

    println("After")
}
// Before
// (1초 후)
// After

위 코드는 코틀린 코루틴 라이브러리에서 delay가 구현된 방식이랑 일치합니다.

현재 delay의 구현은 테스트를 지원하기 위한 목적 때문에 복잡해졌지만, 핵심적인 코드는 거의 같습니다.

 

 

값으로 재개하기

 

resume 함수의 인자와 suspendCoroutine의 타입 인자로 Unit을 사용하고 있다는 것을 보셨을겁니다.

Unit은 함수의 리턴 타입이며, Continuation의 제네릭 타입 인자이기도 합니다.

val ret: Unit =
    suspendCoroutine<Unit> { cont: Continuation<Unit> ->
        cont.resume(Unit)
    }

suspendCoroutine을 호출할 때 Continuation 객체로 반환될 값의 타입을 지정할 수 있습니다.

resume을 통해 반환되는 값은 반드시 지정된 타입과 같은 타입이어야 합니다.

이는 보통 API 통신을 통해 데이터를 받을 때 사용합니다.

 

1. 스레드는 특정 데이터가 필요한 지점까지 비즈니스 로직을 수행한다.
2. 네트워크 라이브러리를 통해 데이터를 요청한다.
3. 코루틴은 데이터를 받고 나면 받은 데이터를 resume을 통해 보낸다.
4. 그 동안 스레드는 다른 일을 할 수 있다.
5. 데이터가 도착하면 중단된 시점에서 재게하게 된다.

suspend fun requestUser(): User {
    return suspendCoroutine<User> { cont ->
        requestUser { user ->
            cont.resume(user)
        }
    }
}

suspend fun main() {
    println("Before")
    val user = requestUser()
    println(user)
    println("After")
}

 

 

예외로 재개하기

 

API가 항상 정상적인 데이터만을 응답하지 않는다는 것을 주의해야합니다.

이런 경우 데이터를 반환할 수 없으므로 예외로 재개하는 방법이 필요합니다.

resume이 호출될 때 suspendCoroutine은 인자로 들어온 데이터를 반환하지만, resumeWithException이 호출되면 중단된 지점에서 인자로 넣어준 예외를 던집니다.

 

class MyException : Throwable("Just an exception")

suspend fun main() {
    try {
        suspendCoroutine<Unit> { cont ->
            cont.resumeWithException(MyException())
        }
    } catch (e: MyException) {
        println("Caught!")
    }
}
// Caught!

이러한 방법은 네트워크 관련 예외를 알릴 때 사용합니다.

suspend fun requestUser(): User {
    return suspendCancellableCoroutine<User> { cont ->
        requestUser { resp ->
            if (resp.isSuccessful) {
                cont.resume(user)
            } else {
                val e = ApiException(
                    resp.code,
                    resp.message
                )
                cont.resumeWithException(e)
            }
        }
    }
}

suspend fun requestNews(): News {
    return suspendCancellableCoroutine<News> { cont ->
        requestNews(
            onSuccess = { news -> cont.resume(news) },
            onError = { e -> cont.resumeWithException(e) }
        )
    }
}