[ Kotlin & Spring ] - 스프링 WebFlux의 코루틴 지원과 코루틴 원리

1. 스프링 WebFLux의 코루틴 지원

프로젝트 리액터 기반의 리액티브 프로그래밍은 비동기-논블로킹의 단점인 콜백 헬 문제를 순차적으로 동작하는 연산자를 통해 해결합니다. 하지만 함수형 패러다임에 익숙하거나 리액터의 다양한 연산자에 부담이 없다면 이도 상관없지만, 매우 러닝커브가 높은 편입니다. 이로 인해 최근 안드로이드도 최근엔 RxJava에서 코루틴 기반으로 작성하는 코드가 늘어나고 있고 서버 측에서도 코루틴을 도입하는 사례도 많아지고 있습니다.

 

1-1. 코루틴

코루틴(Coroutine)은 코틀린에서 비동기-논블로킹 프로그래밍을 명령형 스타일로 작성할 수 있도록 도와주는 라이브러리 입니다. 이는 멀티 플랫폼을 지원하며 코틀린을 사용하는 안드로이드, 서버 등 여러 환경에서 사용할 수 있습니다. 또한 이는 일시 중단 가능한 함수(suspend function)를 통해 스레드가 실행을 잠시 중단했다가 중단한 시점부터 재개(resume)할 수 있습니다!

 

suspend fun combineApi() = coroutineScope {
  val response1 = async { getApi1() }
  val response2 = async { getApi2() }
  return ApiResult (
    response1.await()
    response2.await()
) }

 

위처럼 코루틴을 사용해 구조적 동시성을 구현할 수 있습니다! 여기서 구조적 동시성이란 동시에 실행이 되는 작업이 "구조적"이여야 한다는 것입니다. 코루틴을 사용하는 구조적 동시성에서는, 각 코루틴이 어떤 함수나 클래스의 범위 내에서 생성되고, 해당 범위가 종료될 때 모든 코루틴이 완료되도록 보장합니다. 이렇게 하면 비동기 코드를 더 쉽게 이해하고, 버그를 방지하며, 리소스를 더 효과적으로 관리할 수 있게 됩니다.

 

1-2. 스프링 WebFlux의 코루틴 지원

스프링 WebFlux 공식문서의 코틀린 예제를 보면 모두 코루틴 기반의 예제를 소개하고 있습니다. 또한 스프링 MVC, WebFlux 모두 코루틴을 지원하여 의존성만 추가하면 바로 사용가능합니다. 아래는 리액티브가 코루틴으로 변환되는 형식을 보여줍니다.

 

//Mono → suspend
fun handler(): Mono<Void> -> suspend fun handler()
//Flux → Flow
fun handler(): Flux<T> -> fun handler(): Flow<T>

 

기존의 컨트롤러를 아래와 같이 바꿔볼 수도 있겠습니다.

@RestController
class UserController(
  private val userService : UserService,
private val userDetailService: UserDetailService ){
  @GetMapping("/{id}")
  suspend fun get(@PathVariable id: Long) : User {
    return userService.getById(id)
  }
  @GetMapping("/users")
  suspend fun gets() = withContext(Dispatchers.IO) {
      val usersDeffered = async { userService.gets() }
      val userDetailsDeffered = async { userDetailService.gets() }
      return UserList(usersDeffered.await(), userDetailsDeffered.await())
  }
}

 

또한 WebClient도 아래와 같이 변경할 수 있습니다.

val client = WebClient.create("https://example.org")
val result = client.get()
        .uri("/persons/{id}", id)
        .retrieve()
        .awaitBody<Person>()

기존 리액티브 코드를 코루틴으로 변경하고 싶다면 awaitXXX 로 시작하는 확장 함수를 사용하면 즉시 코루틴으로 변경할 수 있습니다.

 

Spring Data R2DBC의 ReactiveCrudRepository에서의 코루틴 적용

interface ContentReactiveRepository : ReactiveCrudRepository<Content, Long> {
    fun findByUserId(userId: Long) : Mono<Content>
    fun findAllByUserId(userId: Long): Flux<Content>
}
class ContentService (
  val repository : ContentReactiveRepository
){
  fun findByUserIdMono(userId: Long) : Mono<Content> {
    return repository.findByUserId(userId)
}
  suspend findByUserId (userId: Long) : Content {
      return repository.findByUserId(userId).awaitSingle()
      }
 }

위와같이 awaitSingle을 통해 코루틴으로 변경해서 사용할 수 있습니다. 하지만 아래와 같이 CorotineCrudRepository를 사용하면 awaitXXX 코드 없이 사용도 가능합니다.

 

interface ContentCouroutineRepository : CoroutineCrudRepository<Content, Long> {
    suspend fun findByUserId(userId:Long) : Content?
    fun findAllByUserId(userId: Long): Flow<Content>
}
class ContentService (
  val repository : ContentCouroutineRepository
){
  suspend findByUserId (userId: Long) : Content {
      return repository.findByUserId(userId)
} }

좀있다 볼거지만 suspend 함수 안에서는 그냥 함수, suspend함수를 구조적 동시성을 만족시키기 위해 제한합니다.

 

 

2. 코루틴 기초

2-1. runBlocking

runBlocking은 코루틴을 생성하는 코루틴 빌더라고 할 수 있겠습니다. 이로 감싼 코드는 코루틴 내부의 코드가 수행이 끝날때 까지 스레드가 블로킹됩니다.

 

package com.hyunseo.springwebflux.coroutine

import kotlinx.coroutines.runBlocking

/**
 * @author ihyeonseo
 */

fun main() {
    runBlocking {
        println("Hello")
        println(Thread.currentThread().name)
    }

    println("World")
    println(Thread.currentThread().name)
}

 

위와같이 실행 옵션에 -Dkotlinx.corotines.debug를 붙혀서 스레드 이름을 확인해보면, 수행되는 스레드 이름 뒤에 @coroutine#1이 붙어있는 것을 볼 수 있습니다.

 

일반적으로 코루틴은 스레드를 차단하지 않고 사용해야하므로 runBlocking을 사용하는 것은 좋지 않지만 꼭 사용해야 하는 경우가 있습니다. 예를 들어 테스트코드, 스프링 배치등에서 말이죠

 

2-2. launch

laucn는 스레드 차단 없이 새 코루틴을 시작하고 결과로 job을 반환하는 코루틴 빌더입니다. launch는 결과를 만들어내지 않는 비동기 작업에 적합하기 때문에 인자로 Unit을 반환하는 람다를 인자로 받습니다.

 

package com.hyunseo.springwebflux.coroutine

import kotlinx.coroutines.*
import kotlin.system.measureTimeMillis

/**
 * @author ihyeonseo
 */

fun main() = runBlocking<Unit> {

    val job1: Job = launch {
        val elapsedTime = measureTimeMillis {
            delay(150)
        }
        println("async task-1 ${elapsedTime} ms")
    }

//    job1.cancel()

    val job2: Job = launch {
        val elapsedTime = measureTimeMillis {
            delay(100)
        }

        println("async task-2 ${elapsedTime} ms")
    }

//    println("start task-2")
//    job2.start()
}

위에서 일단 delay()함수는 코루틴 라이브러리에 정의된 일시 중단 함수이며 Thread.sleep()과 유사하지만 현재 스레드를 차단하지 않고 일시 중단 시킵니다. 이때 일시 중단 된 스레드는 코루틴 내에서 다른 일시 중단 함수를 수행합니다. 또한 launch가 반환하는 Job을 사용해 현재 코루틴의 상태를 확인하거나 실행 또는 취소도 가능합니다! 즉 코루틴의 생명주기를 관리할 수 있다는 것입니다.

 

위를 보면 job1.cancel() 을 통해 코루틴을 취소할 수도, launch(start = CoroutineStart.LAZY)를 사용해서 start 함수를 호출하는 시점에 코루틴을 동작시킬 수 있습니다. 

 

2-3. async

async 빌더는 비동기 작업을 통해 결과를 만들어 내는 경우에 적합합니다.

 

package com.hyunseo.springwebflux.coroutine

import kotlinx.coroutines.Deferred
import kotlinx.coroutines.async
import kotlinx.coroutines.delay
import kotlinx.coroutines.runBlocking

/**
 * @author ihyeonseo
 */


fun sum(a: Int, b: Int) = a + b

fun main() = runBlocking<Unit> {
    val result1: Deferred<Int> = async {
        delay(100)
        sum(1, 3)
    }

    println("result1: ${result1.await()}")

    val result2 = async {
        delay(100)
        sum(2, 5)
    }

    println("result2: ${result2.await()}")
}

 

async는 비동기 작업의 결과로 Deferred 라는 특별한 인스턴스를 반환하는데 await이라는 함수를 통해 async로 수행한 비동기 작업의 결과를 받아올 수 있습니다. 자바스크립트나 다른 언어의 async-await은 키워드인 경우가 보통이지만 코틀린의 코루틴은 async-await이 함수인 점이 차이점이라고 할 수 있겠습니다.

 

2-4. suspend 함수

suspend 함수는 코루틴의 핵심 요소로써 일시 중단이 가능한 함수를 말합니다. suspend 함수는 일반 함수를 마음껏 호출할 수 있지만 일반 함수에서는 suspend 함수를 호출할 수 없다는 것이 큰 특징입니다. suspend 함수에서 앞서 학습한 async, launch와 같은 코루틴 빌더를 사용하려면 코루틴 스코프(CoroutineScope를 사용합니다.

 

package com.hyunseo.springwebflux.coroutine

import kotlinx.coroutines.Job
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch


suspend fun main() {
    doSomething()
}

fun printHello() = println("hello")

suspend fun doSomething(): Job = coroutineScope {
    launch {
        delay(200)
        println("world!")
    }

    launch {
        printHello()
    }
}

 

coroutineScope를 사용하면 runBlocking과 다르게 현재 스레드가 차단되지 않고 코루틴이 동작합니다. 위에서는 다 차단되었었죠??

 

2-5. Flow

Flow는 코루틴에서 리액티브 프로그래밍 스타일로 작성할 수 있도록 만들어진 API입니다. 코루틴의 suspend 함수는 단일 값을 비동기로 반환하지만 Flow를 사용하면 여러개의 값을 반환할 수 있습니다.

 

package com.hyunseo.springwebflux.coroutine

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.runBlocking

/**
 * @author ihyeonseo
 */

fun main() = runBlocking<Unit>{
    val flow = simple()

    flow.collect { value -> println(value) }
}

fun simple(): Flow<Int> = flow {
    println("Flow started")

    for (i in 1..3) {
        delay(100)
        emit(i)
    }
}

 

여기서 중요한 점은 리액티브 스트림과 같이 Terminal Operator(최종 연산자) 인 collect를 호출하지 않으면 아무런 일도 일어나지 않는다는 점입니다. 

 

 

https://wooooooak.github.io/kotlin/2019/08/25/%EC%BD%94%ED%8B%80%EB%A6%B0-%EC%BD%94%EB%A3%A8%ED%8B%B4-%EA%B0%9C%EB%85%90-%EC%9D%B5%ED%9E%88%EA%B8%B0/

코틀린 코루틴(coroutine) 개념 익히기 · 쾌락코딩

이 블로그에 코루틴의 핵심 개념이 정말 잘 되어 있습니다. 잠깐만 정리해보자면

 

일반적인 상황

우리가 흔히 아는 메인 루틴과 서브 루틴의 동작 방식은 다음과 같습니다. 메인 쓰레드가 plusOne이라는 서브 루틴에 진입합니다. 당연히 코드는 맨 위에서 부터 실행될건데, return 문을 만나면 서브루틴을 호출했던 부분으로 탈출하게 됩니다. 그리고 진입점과 탈출점 사이에 쓰레드는 블럭되어 있습니다. 

 

코루틴을 사용하는 상황

코루틴도 하나의 routine이라고 생각합시다. 그런데 이 함수에 진입할 수 있는 진입점도 여러개고, 함수를 빠져나갈 수 있는 탈출점도 여러개입니다. 즉, 코루틴은 return 문이나 마지막 닫는 괄호를 만나지 않더라도 중간에 언제든지 나갈 수 있고 언제든지 나갔던 그 시점으로 들어올 수 있습니다.

startCorotuine을 가상의 코루틴 빌더 함수라고 가정해봅시다. 그리고 그 안에 여러개의 suspend 함수가 존재하는데, suspend로 선언된 함수를 만나면 코루틴 밖으로 잠시 나갈 수 있게 됩니다. 간단히 순서를 정리하자면

 

1. 쓰레드의 Main 함수가 drawPerson()을 호출하면 해당 쓰레드가 startCoroutine 블럭을 만나 코루틴이 됩니다. 즉 drawPerson은 진입점과 탈출점이 여러개가 되는 자격이 주어진 것입니다.
2. drawHead() 는 suspend 키워드로 정의됩니다. 그래서 drawHead() 부분에서 더 이상 아래 코드를 실행하지 않고 drawperson() 이라는 코루틴 함수를 (잠시) 탈출합니다.
3. 메인 쓰레드가 해당 코루틴을 탈출했습니다. 그렇다고 쓰레드는 노는 것이 아니라 우리가 짜 놓은 다른 코드들을 실행하고 있을 겁니다. 그리고 head는 어디선가 계속 그려지고 있을 겁니다. drawHead() 라는 suspend를 만나 코루틴을 탈출했지만, drawHead() 함수의 기능은 메인쓰레드에서 동시성 프로그래밍으로 작동하고 있을 수도 있고, 다른 쓰레드에서 돌아가고 있을 수도 있습니다. 그것은 개발자의 자유입니다.
4. 그렇게 메인 쓰레드가 다른 코드를 실행하다가도, drawHead()가 제 역할을 다 끝내면서 다시 아까 탈출했던 drawPerson() 으로 돌아옵니다. 아까 멈추어 놓았던 drawHead() 아래인 drawBody() 부터 재개(resume)됩니다.

 

즉 이렇게 한 쓰레드에서 Context-Switching 되지 않고, 동시성 프로그래밍이 가능하게 되어, 엄청난 효율성을 가지게 될 수 있다는 점이 코루틴의 장점이 되겠습니다!