[당근마켓 밋업 정리] - Kotlin Coroutines 톺아보기

[당근마켓 밋업 정리] - Kotlin Coroutines 톺아보기

해당 자료에 대한 내용은 당근마켓 유튜브를 참고하여 정리한 내용입니다.

참고 자료 : https://youtu.be/eJF60hcz3EU

기본 개념

Synchronous (동기)

• 요청과 결과가 동시에 일어난다는 약속.
  • 요청을 하면 시간이 얼마가 걸리던지 요청한 자리에서 결과가 주어져야 함.
• 요청을 하면 System Call이 끝날때까지 기다렸다 결과물을 가져옴.
• 설계가 매우 간단하고 직관적이지만, 결과가 주어질 때까지 어무것도 못하고 대기해야 함.

Asynchronous (비동기)

• 요청과 결과가 동시에 일어나지 않음.
  • 요청한 그 자리에서 결과가 주어지지 않음.
• 요청을 하면 System Call이 완료되지 않아도 나중에 완료가 되면 그때 결과물을 가져옴.
  • 주로 Callback 함수를 통해 결과물을 가져옴.
• 동기에 비해 코드가 복잡하고 추적과 에러 핸들링이 어렵지만, 결과가 주어지는데 시간이 걸리더라도 다른 작업을 할 수 있어 자원을 효율적으로 사용할 수 있음.

비동기에 대한 고민

다양한 비동기 라이브러리

• 스프링에서는 Spring Reactor
• 일부에서는 Java8의 CompletableFuture
• Hibernate Reactive의 MUTINY

어떤걸 사용해야 하고 어떻게 혼용해서 써야 할까?

Coroutine이 해결사?

• 우수한 가독성
• 에러 핸들링
• 동시성 처리
• Flow
• Channel

Coroutine 소개

주문 생성 동기 코드

fun execute(inputValues: InputValues): Order {
    val (userId, productIds) = inputValues

    // 1. 구매자 조회
    val buyer = userRepository.findUserByIdSync(userId)

    // 2. 주소 조회 및 유효성 체크
    val address = addressRepository.findAddressByUserSync(buyer).last()
    checkValidRegion(address)
    
    // 3. 상품들 조회
    val products = productRepository.findAllProductsByIdsSync(productIds)
    check(products.isNotEmpty())
    
    // 4. 스토어 조회
    val stores = storeRepository.findStoresByProductsSync(products)
    check(stores.isNotEmpty())
    
    // 5. 주문 생성
    val order = orderRepository.createOrderSync(buyer, products, stores, address)
    
    return order
}

비동기 코드로

RxJava3의 Maybe

• Maybe는 Observable의 또 다른 형태.
• 최대 데이터 하나를 발행할 수 있으며 추가로 데이터 없이도 완료할 수도 있음.
• 0 .. 1, Error
• 총 3가지 알림을 보냄.
  • onSuccess: 데이터 하나를 발행함과 동시에 종료
  • onError: 에러가 발생했음을 알림
  • onComplete: 데이터 발행이 완료됐음을 알림

interface UserAsyncRepository {
    fun findUserByIdMaybe(userId: String): Maybe<User>
}

class UserRxRepository: UserAsyncRepository {
    
    override fun findUserByIdMaybe(userId: String): Maybe<User> {
        val user = prepareUser(userId)
        return Maybe.just(user)
            .delay(TIME_DELAY_MS, TimeUnit.MILLISECONDS)
    }
}

JDK9 ReactiveStream (구현 생략)

• Non-Blocking과 BackPressure을 이용하여 비동기 서비스를 할 때 기본이되는 스펙.
• 정해진 개수 만큼 Publish 하고나면 complete 이벤트로 Flow 종료.

interface AddressAsyncRepository {
    fun findAddressByUserAsPublisher(user: User): Flow.Publisher<Address>
}

Reactor의 Flux (구현 생략)

• 0 .. n, Error

interface ProductAsyncRepository {
    fun findAllProductsByIdsAsFlux(productIds: List<String>): Flux<Product>
}

Mutiny의 Multi (구현 생략)

• 0 .. n, Error

interface StoreAsyncRepository {
    fun findStoresByProductsAsMulti(products: List<Product>): Multi<Store>
}

JDK8 CompletableFuture

• complete 되는 시점에 결과 반환

interface OrderAsyncRepository {
    fun createOrderAsFuture(
        buyer: User,
        products: List<Product>,
        stores: List<Store>,
        address: Address0
    ): CompletableFuture<Order>
}

위의 비동기 라이브러리를 혼합해서 사용하면 어떤일이 벌어질까?

Callback 지옥... 이런 느낌이 아닐까...?

subscribe 지옥...

• subscribe는 결과를 얻은 시점에 주어진 subscriber (consumer)를 실행하는 일종의 callback.
• 반환값들이 아래에서 계속 필요하여 subscribe가 중첩.

Coroutine

코루틴화 시킬 수 있는 코드는 suspend라는 키워드가 붙은 함수들에 한 해서 가능.

• Maybe<T>.awaitSingle
• Publisher<T>.awaitLast
• Flow<T>.toList
• CompletableFuture<T>.await

위의 함수들은 모두 확장 함수에 포함됨.

suspend fun execute(inputValues: InputValues): Order {
    val (userId, productIds) = inputValues

    // 1. 구매자 조회
    val buyer = userRepository.findUserByIdAsMaybe(userId).awaitSingle()

    // 2. 주소 조회 및 유효성 체크
    val address = addressRepository.findAddressByUserAsPublisher(buyer).awaitLast()
    checkValidRegion(address)

    // 3. 상품들 조회
    val products = productRepository.findAllProductsByIdsAsFlux(productIds).asFlow.toList()
    check(products.isNotEmpty())

    // 4. 스토어 조회
    val stores = storeRepository.findStoresByProductsAsMulti(products).asFlow.toList()
    check(stores.isNotEmpty())

    // 5. 주문 생성
    val order = orderRepository.createOrderAsFuture(buyer, products, stores, address).await()

    return order
}