Bldev

make X better with code

[코틀린] 코틀린 기초

코루틴(coroutine)은 비동기적으로 실행하려는 코드를 보다 쉽게 작성할 수 있게 도와주는 동시 실행 설계 패턴이다. 코루틴은 중단 가능한(suspendable) 연산, 즉 함수가 특정 시점에 실행을 중단했다가 나중에 다시 시작할 수 있다는 개념이며 코틀린 언어는 비동기 코드 ㅎ작업에 대해 접근 방식 중 하나로서 코루틴이라는 기능을 제공한다. 코루틴은 동시적(concurrent) 코드를 마치 동기적(synchronous) 코드처럼(논블로킹 코드를 마치 블로킹 코드처럼) 명령형 방식으로 작성할 수 있게 함으로써 스레드를 통한 동시성 구현을 보다 쉽게 할 수 있게 도와준다. 따라서 코루틴을 사용하면 동시성 프로그래밍을 절차적인 방식으로 수행할 수 있다. 기존 동시성 프로그래밍 구현에서는 동시적으로 실행할 여러 작업들을 선언 후 서로 결합시키고 각각의 작업에 대한 콜백을 별도로 정의하는 방식이었다면 코루틴에서는 비동기적으로 실행할 작업이 정의된 코루틴 코드 블록들을 절차적으로 나열하기만 하면 된다.

코루틴은 코틀린 언어에서 개발한 새로운 개념은 아니며 고(Go) 언어와 같은 다른 프로그래밍 언어에서도 널리 사용되고 있다. 하지만 코틀린에서 코루틴이 구현되는 방식은 대부분의 기능이 라이브러리에 위임되어 있다는 것이다. 실제로 suspend 키워드 외에 다른 키워드는 언어에 추가되지 않으며 이는 언어 문법의 일부로 async 및 await 키워드가 있는 C#, 자바스크립트와 같은 언어와는 다소 다르다. 코틀린에서 이러한 키워드는 라이브러리 함수일 뿐이다.

코루틴은 중단 가능한 연산의 인스턴스이다. 코루틴은 코드 블록을 동시적으로 실행한다는 의미에서 개념적으로 스레드와 유사하지만 특정 코루틴이 특정 스레드와 연관되어 있지는 않다. 코루틴의 코드 블록은 하나의 스레드에서 실행되다가 중단될 수 있고 다른 스레드에서 재개될 수 있다. 따라서 코루틴은 중단 함수(suspend function)로 구성되어야 한다는 제약이 있다.

여러 작업들을 비동기적으로 처리하기 위해 다수의 스레드를 사용하여 동시 실행을 구현하는 대신 코루틴을 사용하면 단일 스레드에서 동시에 여러 작업들을 수행하는 비동기 프로그래밍이 가능하다.

코루틴은 기본 스레드를 블로킹하여 프로그램이 응답하지 않게 만들 수도 있는 블로킹 작업을 관리하는 방법을 제공한다. 실행 중인 스레드를 블로킹하지 않는 중단(suspend) 기능을 지원하므로 단일 스레드에서 많은 코루틴 코드 블록을 실행할 수 있다. 실행 스레드가 요청한 블로킹 작업에 의해 코드 실행이 중지될 수 있지만 실행 스레드는 블로킹되지 않으며 다른 작업을 수행할 수 있다. 하지만 이러한 동작이 항상 보장되는 것은 아니며 이를 위해 올바른 코드 작성이 필요하다.

코틀린의 중단 기능은 많은 동시 작업을 지원하며, 스레드가 블로킹되는 경우 보다 리소스를 절약한다. 코루틴은 기존 스레드 사용 방식이 비해 메모리 사용 및 컨텍스트 스위칭 수행 시간 관점에서 오버헤드가 낮다. 코루틴은 추상화된 고레벨 API를 기반으로 동작하여 스레드 간 공유 객체 잠금 및 동기화를 관리할 필요가 없어 기존 스레드보다 사용하기 쉽다.

코루틴은 중단할 수 있다. 코루틴의 중단 가능한 특성은 코루틴 코드 블록이 실행되던 스레드를 블로킹하지 않고 다른 코루틴의 코드를 해당 스레드에서 실행될 수 있도록 한다. 다음은 코루틴 코드 실행의 예이다.

runBlocking { // 외부 코루틴 스코프를 생성한다.
  launch { // 새로운 내부 코루틴 스코프를 생성하고 코루틴을 실행한다.
    // 논블로킹 지연에 의해 코루틴이 실행되다가 중지된다. 스레드는 블로킹되지 않는다.
    delay(1000L) // 3
    // 지연 후에 중지된 코루틴이 스레드 상에서 재개된다.
    println("World!") // 5
  }
  
  // 내부 코루틴이 실행 완료되지 않아도 외부 코루틴이 스레드 상에서 실행된다.
  // 내부 코루틴이 지연되는 동안 메인 코루틴이 실행된다.
  println("Hello") // 1
  
  launch { // 새로운 내부 코루틴 스코프를 생성하고 코루틴을 실행한다.
    delay(2000L) // 4
    println("World2!") // 6
  }
  
  println("Hello2") // 2
}


스레드가 블로킹되지 않고 다른 작업을 수행할 수 있는 비동기, 논블로킹 방식에서는 하나의 스레드가 여러 작업을 번갈아 수행하여 리소스 효율을 높일 수 있다. 필요에 따라 CPU 연산 집약적 작업이나 IO 작업을 별도의 스레드에게 할당할 수도 있다.

메인 스레드가 프로그램 실행 도중 다른 블로킹 작업에 의해 방해를 받으면 안 되는 경우 해당 작업을 별도의 스레드에게 요청하는 것이 필요하다. 메인 스레드는 요청한 작업이 완료될 때까지 블로킹되지 않고 다른 작업을 수행한다. 예를 들어, UI 처리를 담당하는 전용 스레드는 사용자의 입출력 작업을 위해 블로킹되서는 안 된다. 코루틴의 중단 기능은 이를 손쉽게 가능하게 한다.


코루틴 라이브러리

코틀린은 다른 라이브러리에서 코루틴을 활용할 수 있도록 표준 라이브러리인 kotlinx.coroutines를 통해 최소한의 저수준 API만 제공한다. 유사한 기능을 가진 다른 많은 언어와 달리 async 및 await는 코틀린의 키워드가 아니며 표준 라이브러리의 일부도 아니다. 코틀린의 함수 일시 중단 개념은 퓨처(future)와 프로미스(promise)보다 더 안전하고 오류가 덜 발생하는 비동기 연산에 대한 추상화를 제공한다.

kotlinx.coroutines 라이브러리에는 여러 가지 고수준의 코루틴 지원 프리미티브가 포함되어 있다. 코루틴을 사용하려면 org.jetbrains.kotlinx:kotlinx-coroutines-core 종속성을 추가해야 한다.

kotlinx.coroutines 라이브러리의 코어 모듈은 멀티플랫폼도 지원한다. JVM 뿐만 아니라 다양한 운영 체제에서 동작하는 코틀린/네이티브(Kotlin/Native), 자바스크립트를 위한 코틀린/자바스크립(Kotlin/JS)에서도 kotlinx.coroutines 라이브러리의 코어 모듈을 사용하여 코루틴 기능을 사용할 수 있다.


중단 함수

suspend 키워드로 정의한 함수는 실행 도중 중지가 가능한 함수이다. 해당 함수를 실행한 스레드에서 실행 도중 임시로 중지하고 나중에 다른 스레드에서 중지된 함수를 다시 재개할 수 있다. 이러한 함수를 중단 함수(suspend function)라고 한다.

중단 함수는 코루틴과 함께 사용되어 비동기 및 논블로킹 작업을 수행하는데 사용된다. 이는 복잡한 멀티스레딩 구현 필요 없이 비동기, 논블로킹 방식으로 하나의 작업을 여러 스레드가 실행할 수 있는 방법을 제공한다. 따라서 중단 함수를 사용하여 메인 스레드를 블로킹하지 않고 오랜 시간이 걸리는 작업을 수행할 수 있다.

중단 함수의 실행이 중단되었을 때 작업을 수행하던 스레드는 다른 작업을 수행할 수 있다. 중단된 이후 작업은 가용 스레드에 의해 다시 재개된다.

중단 함수는 코루틴 내부에서 일반 함수처럼 사용될 수 있다. 일반 함수와 달리 중단 함수는 코루틴의 실행을 중지하기 위해 내부적으로 다른 중단 함수(예: delay 함수)를 사용할 수 있다는 특징이 있다.

중단 함수는 반드시 중단 함수 또는 코루틴 내에서 호출되어야 한다. 따라서 코루틴을 사용하여 중단 함수를 비동기로 실행하기 위해서 코루틴 빌더(또는 코루틴 스코프 빌더)를 사용하여 코루틴 스코프 정의가 먼저 필요하다. 코루틴 내에서는 일반 함수와 중단 함수 모두 호출 가능하다.

중단 함수 내에서 스레드를 블로킹할 수 있는 작업(IO 작업, CPU 집약적 연산 작업)들을 단순히 순차적으로 호출한다고 해서 스레드 블로킹이 방지되는 것이 아니다. 중단 함수는 비동기 작업을 쉽게 처리할 수 있도록 설계되었지만, 블로킹 연산에 의한 스레드 블로킹을 막지는 못한다. 중단 함수 내 블록은 논블로킹 컨텍스트(non-blocking context)이다. 중단 함수에서 실행할 작업들을 코루틴으로 정의하고 코루틴 실행에 사용되는 스레드를 결정하는 코루틴 디스패처(dispatcher)가 사용되었을 때 중단 함수는 실행 중인 현재 스레드를 블로킹하지 않고 특정 스레드에서 블로킹 작업을 수행하도록 하고, 현재 스레드에서 다른 작업이 수행될 수 있도록 위임한다. 논블로킹 컨텍스트에서 디스패처 사용 없이 블로킹 호출을 수행하는 경우 블로킹 호출은 실행 중인 스레드를 블로킹하여 다른 작업이 해당 스레드에서 수행될 수 없게 만든다. 결과적으로 가용 스레드가 감소하고 스레드 풀이 고갈된다.

젯브레인스가 제공하는 Inspectopedia링크는 코드 검사(inspection) 도구이다. 코드베이스의 문제를 식별하고 그 이유에 대한 자세한 정보를 제공하여 개발자가 코드 품질을 향상시키고 보안 문제나 성능 문제를 조기에 발견하여 수정할 수 있도록 지원한다. 중단 함수에서 블로킹 호출을 수행하는 코드를 작성하는 경우 Inspectopedia는 Possibly blocking call in non-blocking context 경고를 발생시킨다링크. 스레드 블로킹을 막기 위해 블로킹 작업은 호출 스레드와 다른 스레드에서 수행되어야 하며 이를 위해 withContext() 코루틴 확장 함수를 사용하거나 블로킹 코드를 대체할 수 있는 논블로킹 코드를 사용하는 것이 권장된다. withContext() 함수는 특정 디스패처를 사용하여 블로킹 코드를 현재 스레드가 아닌 별도의 스레드 상에서 실행될 수 있도록 하여 현재 스레드가 블로킹 되지 않게 한다.

중단 함수 내에 정의된 코루틴은 중단 함수가 중단되는 지점을 제공한다. 중단 함수는 코루틴이 실행되는 동안 중단되며 코루틴에서 값을 반환하거나 코루틴에서 호출한 중단 함수가 결과를 반환할 때 코루틴이 완료된다.

suspend fun loadData() {
  // 외부 중단 함수 실행이 중단된다.
  // 외부 코루틴이 완료되면 중단된 외부 함수가 계속 실행된다.
  val data = coroutineScope {
    fetchData()
  } 
  
  // 내부 중단 함수 실행이 완료되어 외부 코루틴이 완료되면 실행된다.
  return data
}

중단 함수 내에서 코루틴을 실행하고 그 결과를 기다리는 동안 현재 스레드를 블로킹되지 않으며 스레드는 다른 작업을 수행할 수 있다.

부모 코루틴 스코프 내에서 정의된 모든 자식 코루틴이 완료되도록 보장하는 것이 중요하다. 하나 이상의 자식 코루틴을 실행하는 부모 코루틴 스코프를 정의한 후 단일 코루틴일 경우 await(), 여러 코루틴일 경우 awaitAll()을 사용하여 함수가 값을 반환하기 전에 모든 자식 코루틴이 완료되도록 보장할 수 있다.


코루틴 스코프와 코루틴 컨텍스트

코루틴은 구조적 동시성(structured concurrency) 원칙을 따른다. 이는 코루틴은 특정 코루틴 스코프(영역) 내에서만 실행될 수 있다는 원칙이다. 코루틴 스코프는 하나 이상의 관련 코루틴의 생명주기를 제한 및 관리한다.

새로운 코루틴 스코프를 생성하기 위해서는 CoroutineScope 인터페이스를 사용한다. 코루틴 스코프를 생성한 후 코루틴 스코프 내에 코루틴을 정의한다. 모든 코루틴은 코루틴 스코프 내에서 실행되어야 한다. 모든 코루틴은 항상 코루틴 스코프와 연결되어 있다.

CoroutineScope 인스턴스는 CoroutineScope() 또는MainScope() 팩토리 함수를 통해 생성할 수 있다. 코루틴 스코프의 독립형 인스턴스를 얻는 가장 좋은 방법은 두 팩토리 함수를 사용하는 것이다. 코루틴 스코프가 더 이상 필요하지 않을 때 이를 취소하는데 주의해야 한다. + 연산자를 사용하여 추가적인 컨텍스트 요소를 스코프에 추가할 수 있다. CoroutineScope 인터페이스를 직접 구현하는 것은 권장되지 않으며, 대신 델리게이션(delegation) 의한 구현이 권장된다.

CoroutineScope()는 일반적인 목적의 스코프를 생성한다. CoroutineScope()는 제공된 컨텍스트를 코루틴의 파라미터로 사용하며 Job이 컨텍스트의 일부로 제공되지 않은 경우 Job을 추가한다. MainScope()는 UI 애플리케이션을 위한 스코프를 생성하며 Dispatchers.Main을 기본 디스패처로 사용한다. MainScope()는 SupervisorJob을 갖고 있다.

CoroutineScope()의 커스텀한 사용의 핵심은 수명 주기가 끝날 때 코루틴 스코프를 취소하는 것이다. 코루틴을 실행하던 엔티티가 더 이상 필요하지 않다면 CoroutineScope의 cancel() 확장 함수를 사용해야 한다. 이 함수는 아직 실행 중일 수 있는 모든 코루틴을 취소한다.

모든 코루틴 빌더(launch, async 등)와 모든 스코핑 함수(coroutineScope, withContext 등)는 실행하는 내부 코드 블록에 고유한 잡(job) 인스턴스와 함께 자체 스코프를 제공한다. 스코핑 함수는 블록 내의 모든 코루틴이 완료될 때까지 기다렸다가 스스로 완료되므로 구조적 동시성을 강제한다.

모든 코루틴은 CoroutineContext라는 코루틴 컨텍스트(코루틴 실행 환경) 내에서 실행된다. 코루틴 컨텍스트는 코루틴 간에 상태를 공유하기 위해 사용된다. 코루틴 컨텍스트는 다양한 요소들의 집합이다. 주요 요소로는 잡과 디스패처가 있다. 코루틴 스코프를 정의하는 모든 코루틴 빌더는 CoroutineContext를 상속하여 컨텍스트의 모든 요소들과 코루틴 취소를 자동으로 전파한다.


잡과 코루틴의 생명주기

Job 인터페이스는 취소 가능한 백그라운드 작업이며 완료(complete)라는 절정의 상태로 끝나는 생명주기를 가진다. Job은 부모-자식의 계층 구조 형태로 구성될 수 있으며, 이때 재귀적으로 부모 Job의 실패는 즉시 모든 자식 Job들의 실패로 이어진다. CancellationException가 아닌 예외와 함께 자식이 실패하면 부모가 즉시 취소되고 결과적으로 모든 자식(동등 레벨의 자식 및 하위 레벨의 자식)들도 취소된다. 이러한 과정은 Job의 기본적인 설계이며 Job 대신 SupervisorJob를 사용하여 이러한 과정을 사용자화할 수도 있다.

모든 코루틴은 Job 인스턴스를 가지고 있다. Job은 생성되는 코루틴들을 식별하고 자체 생명주기를 관리하는데 사용되는, 코루틴 별로 고유한 객체이다. 코루틴 스코프의 컨텍스트는 코루틴 취소 전파와 구조적 동시성 원칙을 강제하기 위해 Job 인스턴스를 포함해야 한다. 코루틴 스코프 컨텍스트의 요소인 Job은 코루틴의 상태 및 생명주기 관리와 관련이 있다.

Job 인터페이스의 가장 기본적인 인스턴스는 다음과 같이 생성된다.

  • 코루틴 잡은 launch 코루틴 빌더로 생성된다. launch 코루틴 빌더는 지정된 코드 블록을 실행하고 이 블록이 완료되면 완료된다.
  • complete() 함수를 통해 완료될 수 있는 잡인 CompletableJob은 Job() 팩토리 함수로 생성된다.


개념적으로 잡의 실행은 결과값을 생성하지 않으며 부수 효과(side effect)를 위해서만 실행된다. 결과값을 생성하는 잡은 Deferred 인터페이스이다.

구조적 동시성은 프로그램 내에 존재하는 수많은 코루틴들이 손실되거나 누수되지 않도록 한다. 부모 코루틴 스코프는 스코프 내에서 실행되는 모든 자식 코루틴이 실행 완료되기 전까지는 완료될 수 없다. 즉, 스코프 내 모든 코루틴이 완료되어야 부모 코루틴 스코프도 완료된다. 구조적 동시성은 코드 실행 중 발생하는 모든 에러가 코루틴 스코프 내에서 적절히 발생하고 사라지지 않도록 한다.

코루틴 빌더를 사용한 코루틴 스코프 설정은 코루틴의 생명주기를 관리하는데 사용된다. 코루틴의 생명주기란 코루틴 생성, 코루틴 실행, 코루틴 중지 및 재개, 코루틴 실행 완료 및 취소, 코루틴 종료와 같이 코루틴이 처리되는 단계를 말한다. runBlocking과 coroutineScope 스코핑 함수는 코루틴 스코프 내의 모든 코루틴(모든 자식 코루틴)이 완료될 때까지 기다리고, 만약 코루틴이 하나라도 실패하면 나머지 코루틴을 취소한다.


supervisorScope를 통한 부모 코루틴과 자식 코루틴 관리

supervisorScope는 코루틴 빌더 중 하나이다. coroutineScope 빌더와 달리 supervisorScope 빌더를 부모 코루틴 빌더로 사용하면 부모 코루틴의 실패가 자식 코루틴에 전파되지 않게 할 수 있다.

supervisorScope는 SupervisorJob으로 CoroutineScope를 만들고 이 스코프로 지정된 중지 블록을 호출한다. 해당 스코프의 외부 스코프에서 coroutineContext를 상속하지만 SupervisorJob으로 컨텍스트의 Job을 재정의한다. 이 함수는 주어진 블록과 모든 하위 코루틴이 완료되는 즉시 반환된다.


Job vs. Deffered

Job은 코루틴의 상태(실행, 완료, 취소)를 추적 및 관리하는데 사용되는 객체이며 코루틴의 구조적 동시성과 관련이 있다. Job을 사용하여 실행 중인 코루틴을 취소하거나 코루틴이 완료된 이후에 나머지 코드를 실행하기 위해 코루틴 실행 완료를 기다리게 할 수 있다. 구조적 동시성에 의해 부모 Job이 취소되면 관련된 모든 자식 Job이 취소된다.

Deferred는 코루틴의 실행 완료 결과를 받기 위해 사용된다. Deferred는 Job을 상속하므로 Job의 목적과 기능을 동일하게 갖는다. Deferred는 결과를 포함하는 Job이다.


코루틴 빌더

코루틴을 정의하기 위해 먼저 코루틴 스코프를 정의해야 한다. 코루틴 스코프는 코루틴 빌더(coroutine builder)를 사용하여 정의한다. 코루틴 빌더는 새로운 코루틴의 스코프를 정의한다. 모든 코루틴 빌더는 CoroutineScope 인터페이스의 확장(extension)이며, coroutineContext를 상속하여 컨텍스트의 모든 요소와 코루틴 취소를 자동으로 전파한다. 코루틴 빌더는 코루틴의 실행 및 관리를 위한 코루틴 컨텍스트를 제공하는 역할을 한다.

코루틴 빌더에는 runBlocking, coroutineScope, async, launch 빌더가 있다. 이러한 코루틴 빌더를 사용하여 코루틴을 실행할 스코프를 지정한다.

코루틴 빌더를 통해 코루틴 스코프를 정의하고 실행할 코루틴을 포함시킨다. 해당 코루틴 스코프 내에서 개별 작업을 비동기적으로 수행하기 위해 launch 또는 async 빌더를 사용할 수 있다. launch와 async 빌더는 새로운 코루틴을 시작함과 동시에 나머지 코드를 실행한다. 코루틴 다음의 나머지 코드(코루틴 스코프 외부 코드)는 코루틴과 독립적으로 실행된다. 이를 통해 코루틴 코드 블록을 비동기, 논블로킹 방식으로 실행하는 함수를 정의할 수 있다. runBlocking 빌더는 최상위 함수인 반면 launch와 async 빌더는 CoroutineScope의 확장 함수이다. 따라서 launch와 async 빌더는 오직 CoroutineScope 내에서만 선언될 수 있다. async, launch 빌더는 코루틴을 생성하고 함수 본문을 실행한다.

runBlocking과 coroutineScope 빌더의 경우 모든 자식 코루틴이 실행 완료되기 전까지 코루틴 스코프가 완료되지 않는 점은 동일하지만 다음과 같은 차이점이 있다. runBlocking 빌더는 내부 코루틴이 실행 완료될 때까지 현재 스레드를 블로킹하는 반면(코루틴 다음 코드 실행 흐름이 블로킹된다), coroutineScope 빌더는 내부 코루틴들을 실행하다가 중단하고 현재 스레드를 블로킹하는 대신 릴리즈시켜 다른 코루틴 실행에 사용될 수 있게 한다. 이러한 특성으로 인해 runBlocking 빌더는 일반 함수이며, coroutineScope 빌더는 중단 함수이다.

코루틴 빌더의 종류와 설명은 다음과 같다.


  1. runBlocking 빌더

runBlocking 빌더는 코루틴 스코프를 만드는 빌더이다. 실행된 모든 자식 코루틴(코루틴 스코프 내 정의된 모든 코루틴)이 완료될 때까지 기다리는 일반 함수이다. 모든 자식 코루틴이 완료될 때까지 현재 스레드를 블로킹한다. runBlocking 빌더는 다른 코루틴 빌더와 달리 중단 함수가 아니므로 일반 함수에서 호출할 수 있다. 따라서 runBlocking 빌더는 코루틴이 아닌 일반 함수의 코드를 runBlocking 빌더 내부의 코루틴과 연결한다. 즉, 코루틴이 아닌 코드에서 코루틴을 실행할 때 사용된다. runBlocking 빌더는 중단 함수를 인자로 전달 받아 해당 함수를 실행하고, 실행한 함수가 반환하는 값을 반환한다.

runBlocking 빌더를 사용하여 코루틴 작업을 실행하는 함수는 다음과 같이 정의한다.

fun 함수명(): 반환타입 {
  return runBlocking {
    ...
  }
}


코루틴에서 실행되는 작업이 값을 반환하는 경우 함수의 반환 타입을 명시할 수 있다. 값을 반환하지 않는다면 Unit 타입을 명시한다. 일반 함수 대신 중단 함수로 정의도 가능하지만 권장되지 않는다. 함수의 반환 타입을 명시하는 대신 빌더를 직접 함수에 할당할 수도 있다.

fun 함수명() = runBlocking {
  ...
}


runBlocking 빌더는 애플리케이션의 최상위 수준에서만 사용되며 실제 코드 내부에서는 거의 사용되지 않는 경우가 많다. 스레드는 고가의 리소스이며 스레드를 블로킹하는 것은 비효율적이고 의도되지 않는 경우가 많기 때문이다. runBlocking 빌더는 이러한 스레드를 블로킹하는 특성으로 인해 적은 수의 스레드를 사용하여 동작하는 코드를 구현하는 리액티브 프로그래밍에서는 성능 저하를 일으킬 수 있다. 이로 인해 실제 서비스 코드에서 사용하는 대신 명령줄 검증 또는 테스트 시 사용하는 것이 좋다.

중단 함수에서 runBlocking 빌더를 통해 코루틴을 호출할 수도 있지만 이러한 구현은 올바르지 않다. 예를 들어 다음 코드는 올바르지 않다.

suspend fun loadData() {
  // 스레드를 블로킹한다.
  val data = runBlocking {
    // 중단 함수
    fetchData()
  }
}


위와 같은 코드에서는 내부 중단 함수가 중단되면 외부 중단 함수가 실행되는 스레드가 해제되는 대신 블로킹되어 잠재적으로 스레드가 고갈되어 성능 문제가 발생할 수 있다. 중단 함수와 같이 중단 가능한 컨텍스트에서 현재 스레드를 블로킹하면 불필요하게 현재 스레드를 블로킹하게 된다. 이는 코루틴의 장점인 비동기성과 스레드 논블로킹을 활용하지 못하며 오히려 성능을 저해한다.

중단 함수의 블록은 중단 가능한 컨텍스트이므로 이 곳에서 runBlocking 빌더를 사용하면 불필요하게 현재 스레드를 블로킹하게 된다. 따라서 중단 함수 내에서는 runBlocking 빌더 대신 coroutineScope 빌더를 사용하여 코루틴을 구성하는 것이 권장된다.

runBlocking 빌더를 사용한 코루틴 실행 코드 예는 다음과 같다.

fun runBlockingFun(): Data {
  작업1 (일반 함수)
  
  // 코루틴이 실행 완료될 때까지 현재 스레드가 블로킹된다.
  val data: Data = runBlocking {
    작업2 (일반 함수 또는 중단 함수)
  }
  
  return data
}

fun runBlockingFun(): Data {
  작업1 (일반 함수)
  
  // 코루틴이 실행 완료될 때까지 현재 스레드가 블로킹된다.
  val data: Data = runBlocking {
    launch {
      작업2-1 (일반 함수 또는 중단 함수)
    }
  
    launch {
      작업2-2 (일반 함수 또는 중단 함수)
    }
  
    작업3 (일반 함수 또는 중단 함수)
  }
  
  return data
}

fun runBlockingFun() = runBlocking {
  작업1 (일반 함수 또는 중단 함수)
  작업2 (일반 함수 또는 중단 함수)
}

fun runBlockingFun() = runBlocking {
  작업1 (일반 함수)
  
  // 코루틴이 실행 완료될 때까지 현재 스레드가 블로킹된다.
  val data: Data = runBlocking {
    launch {
      작업2-1 (일반 함수 또는 중단 함수)
    }
  
    launch {
      작업2-2 (일반 함수 또는 중단 함수)
    }
  
    작업3 (일반 함수 또는 중단 함수)
  }
  
  return data
}


  1. coroutineScope 빌더

coroutineScope 빌더는 코루틴 스코프를 만드는 또하나의 빌더이다. runBlocking 빌더와 동일하게 실행된 모든 자식 코루틴이 완료되기 전까지는 스코프를 완료하지 않는다. 실행된 모든 자식 코루틴이 완료될 때까지 기다리는 중단 함수이다. coroutineScope 빌더는 중단 함수 내에서 여러 개의 동시 작업을 수행하는데 사용된다.

coroutineScope 빌더를 사용하여 코루틴 작업을 실행하는 함수는 다음과 같이 정의한다.

suspend fun 함수명(): 반환타입 {
  return coroutineScope {
    ...
  }
}


코루틴에서 실행되는 작업이 값을 반환하는 경우 함수의 반환 타입을 명시할 수 있다. 값을 반환하지 않는다면 Unit 타입을 명시한다. 일반 함수는 사용할 수 없으며 중단 함수로만 정의가 가능하다. 함수의 반환 타입을 명시하는 대신 빌더를 직접 함수에 할당할 수도 있다.

suspend fun 함수명() = coroutineScope {
  ...
}


runBlocking 빌더와 달리 자식 코루틴이 모두 완료될 때까지 현재 스레드를 블로킹하지 않는다. 현재 스레드를 블로킹하는 대신 일시 중단하여 기본 스레드를 해제한다. 중단 함수이기 때문에 반드시 중단 함수 또는 다른 코루틴 내에서 호출되어야 한다. 즉, 반드시 중단 함수의 일부로 호출되어야 한다.

coroutineScope 빌더로 정의한 중단 함수를 코루틴(부모 코루틴)에서 호출하는 경우 함수 호출 시 부모 코루틴이 중지된다. 이때 스레드는 블로킹되지 않는다. 부모 코루틴을 정의하는 coroutineScope는 모든 자식 코루틴이 완료될 때까지 해당 코루틴 중단 함수를 호출한 부모 코루틴을 재개하지 않는다.

coroutineScope 빌더를 사용한 코루틴 실행 코드 예는 다음과 같다.

suspend fun coroutineScopeFun(): Data {
  작업1 (일반 함수 또는 중단 함수)
  
  // 코루틴이 실행 완료될 때까지 현재 스레드가 블로킹되지 않는다.
  val data: Data = coroutineScope {
    작업2 (일반 함수 또는 중단 함수)
  }
  
  return data
}

suspend fun coroutineScopeFun(): Data {
  작업1 (일반 함수 또는 중단 함수)
  
  // 코루틴이 실행 완료될 때까지 현재 스레드가 블로킹되지 않는다.
  val data: Data = coroutineScope {
    launch {
      작업2-1 (일반 함수 또는 중단 함수)
    }
  
    launch {
      작업2-2 (일반 함수 또는 중단 함수)
    }
  
    작업3 (일반 함수 또는 중단 함수)
  }
  
  return data
}

suspend fun coroutineScopeFun() = coroutineScope {
  작업1 (일반 함수 또는 중단 함수)
  작업2 (일반 함수 또는 중단 함수)
}

suspend fun coroutineScopeFun() = coroutineScope {
  작업1 (일반 함수 또는 중단 함수)
  
  launch {
    작업2-1 (일반 함수 또는 중단 함수)
  }
  
  launch {
    작업2-2 (일반 함수 또는 중단 함수)
  }
  
  작업3 (일반 함수 또는 중단 함수)
}


  1. launch 빌더

launch 빌더는 새로운 코루틴을 시작함과 동시에 나머지 코드를 실행한다. 코루틴 다음의 나머지 코드는 코루틴과 독립적으로 실행된다. 파라미터로 전달하는 함수는 중단 함수이고 아무것도 반환하지 않아야 한다. 따라서 실행한 코루틴에서 반환되는 값을 받을 필요가 없다면 launch를 사용한다. 콘솔 출력, UI 갱신과 같은 부수 효과 일으키는 작업을 백그라운드에서 실행하는데 사용될 수 있다.

launch 함수는 코루틴 실행 후 결과를 Job 객체로 반환한다. Job 객체는 코루틴의 상태를 추적 및 관리하는데 사용되는 객체이다. Job 객체를 사용하여 실행 중인 코루틴을 취소하거나 코루틴이 완료된 이후에 나머지 코드를 실행하기 위해 코루틴 실행 완료를 기다리게 할 수 있다.


  1. async 빌더

async 빌더는 launch 빌더와 마찬가지로 새로운 코루틴을 시작함과 동시에 나머지 코드를 실행한다. 코루틴 다음의 나머지 코드는 코루틴과 독립적으로 실행된다. 코루틴 스코프 내에서 새로운 코루틴을 실행하는 중단 함수이다. 동일한 스코프 내 다른 코루틴과 동시에 실행 가능하다. launch 빌더와 달리 파라미터로 전달하는 함수는 중단 함수이고 값을 반환할 수 있다. 따라서 실행한 코루틴에서 반환되는 값을 호출자가 받을 필요가 있다면 launch 빌더 대신 async를 사용한다.

async 빌더는 코루틴 실행 후 결과를 Deferred 객체로 반환한다. Deferred 객체는 코루틴의 실행 완료 결과를 받기 위해 사용된다. Deferred 객체는 Job을 상속한다. Deferred 객체는 결과를 갖는 Job이다.

Deferred 객체의 await() 함수는 async 빌더가 Deferred 객체를 반환할 때까지 코루틴 실행을 일시 중지시킨다. 즉, await() 함수는 코루틴이 생성한 값을 리턴하기 전에 코루틴이 완료될 때까지 기다리도록 한다.

함수 호출 시 await() 함수를 사용하여 코루틴을 비동기로 실행한 후 종료될 때까지 결과를 기다리는 동기적 작업 수행을 할 수 있다. 즉, async, await() 함수를 사용하면 코루틴을 시작하고 코루틴이 완료될 동안 기다리는 동기적 코드 작성이 가능하다. await() 함수를 사용하여 async 빌더의 실행 결과를 동기적으로 기다릴 수 있지만 현재 스레드를 블로킹하지는 않으며 스레드는 다른 작업을 수행할 수 있다.

await() 함수는 현재 작업 스레드가 아닌 코루틴(외부 코루틴, 중단 함수)을 중지한다. 백그라운드 작업이 수행되고 결과를 기다리는 동안 코루틴을 중지한 후 스레드는 다른 작업을 수행할 수 있으며, 요청 작업이 완료되면 코루틴이 다시 재개된다. 코루틴을 재개하는 스레드는 기존 스레드가 아닌 다른 스레드일 수 있다.

async 빌더와 await() 함수를 사용하면 비동기 및 논블로킹 방식으로 코루틴을 시작 및 실행하고 코루틴이 완료될 동안 기다리는(비동기 작업의 결과를 기다리는) 코드 작성이 가능하다. async, await() 코드는 withContext()를 사용하여 간소화할 수 있다.

async와 launch 빌더의 차이점은 다음와 같다.

  • async 빌더는 Deferred 객체를 통해 코루틴 실행 완료 결과를 반환할 수 있지만 launch 빌더는 실행 결과를 반환할 수 없다.
  • async 빌더는 코루틴 실행 중 예외 발생 시 Job 객체를 통해 예외가 전파되지만 launch 빌더를 호출한 호출자까지 전파되지는 않는다.


GlobalScope

GlobalScope는 특정 작업에 구속되지 않는 코루틴 스코프이다. GlobalScope은 전체 애플리케이션 라이프사이클 동안 실행되며 조기에 취소되지 않는 최상위 코루틴을 실행하는데 사용된다.

메인 스레드가 종료될 때 자동으로 종료되는 백그라운드 스레드인 데몬 스레드와 유사하게 프로세스가 종료될 때 GlobalScope에서 시작된 코루틴도 종료된다. 이러한 특성 때문에 GlobalScope는 애플리케이션의 생명주기와 독립적인 장기 실행 작업에 사용될 수 있지만 관리되지 않는 코루틴의 경우 계속적으로 실행되어 메모리와 같은 리소스 누수 문제를 일으킬 수 있으므로 주의해서 사용해야 하는 섬세한(delicate) API이다.

GlobalScope에서 실행되는 코루틴은 구조적 동시성 원칙의 적용을 받지 않으므로 네트워크 속도 저하 등의 문제로 인해 작업 실행이 중단되거나 지연되는 경우에도 코루틴이 계속 실행되면서 리소스가 소모된다. 따라서 대부분의 경우 GlobalScope을 사용하는 대신 다음과 같은 방법을 사용하여 코루틴을 실행하는 것이 좋다.

  • 단일 작업을 실행하는 경우 중단 함수를 사용한다.
  • 여러 작업을 동시적으로 실행하는 경우 구조적 동시성을 제공하는 coroutineScope 빌더를 사용하여 코루틴 실행을 관리한다.


애플리케이션의 전체 라이프사이클 동안 활성 상태를 유지해야 하는 최상위 백그라운드 프로세스와 같이 GlobalScope가 올바르고 안전하게 사용될 수 있는 사례도 있다.


withContext()

withContext()는 CoroutineScope의 확장 함수이며 주어진 코루틴 컨텍스트와 함께 파라미터로 전달한 중단 함수 블록을 호출하고, 코루틴 실행이 완료될 때까지 외부 코루틴을 중지한 후에 그 결과를 반환한다. withContext() 함수 자체도 중단 함수이다.

withContext() 중단 함수 블록의 코드가 실행되기 전에 외부 코루틴은 중지된다. 코드가 실행 완료되면 중지된 외부 코루틴이 재개된다.

파라미터로 전달하는 중단 함수는 취소 가능하다. 호출 취소 시 결과 컨텍스트는 취소를 즉시 확인한 후 해당 코루틴 스코프 내 코루틴 컨텍스트의 Job이 활성화되어 있지 않다면 CancellationExecption이 발생된다.

withContext()의 인자로 코루틴 디스패처(dispatcher)를 전달하는 경우 특정 스레드풀에서 실행되는 코드 블록을 만들며 블록 내의 코드는 항상 디스패처를 통해 특정 스레드에서 실행된다. 함수 정의 시 withContext()를 호출함으로써 해당 함수의 코드 실행을 특정 스레드에서 수행하도록 할 수 있다.

withContext()의 인자로 전달하는 코루틴 컨텍스트가 현재의 컨텍스트와는 다른 코루틴 디스패처(CoroutineDispatcher)를 제공하는 경우 추가적인 디스패치(스레드 이동)를 수행한다. 코루틴 블록은 즉시 실행되지 않으며 전달된 코루틴 디스패처에서 실행되어야 한다. 코루틴 블록이 실행 완료되면 코루틴 실행은 원래의 디스패처로 다시 이동해야 한다.

withContext()를 사용하면 콜백을 도입하지 않고도 코드 실행 스레드풀을 제어할 수 있다. 따라서 데이터베이스 조회 요청, 네트워크 요청과 같은 작업을 수행하는 매우 작은 크기의 함수에 이를 적용할 수 있다. 해당 작업을 작업을 요청한 기본 스레드가 아닌 별도의 스레드에서 실행하도록 함으로써 기본 스레드의 작업 수행을 방해하지 않는 안전한 함수 호출이 가능하다. 네트워크 작업을 수행하는 함수를 기본 스레드에서 호출하게 되면 기본 스레드는 요청 작업이 끝날 때까지 기다리게 된다. 즉, 기본 스레드가 블로킹될 수 있다. 이러한 함수 호출은 안전하지 않다. 대신 디스패처를 사용하여 작업을 기본 스레드가 아닌 별도의 스레드에서 수행되도록 하면 안전한 호출이 가능하다.


디스패처

코루틴은 CoroutineContext라는 컨텍스트 내에서 실행된다. 이 코루틴 컨텍스트에는 CoroutineDispatcher라는 코루틴 디스패처가 존재한다. 코루틴 디스패처는 코루틴을 어떤 스레드 또는 스레드풀에서 실행할지 결정한다. 즉, 코루틴 디스패처를 통해 코루틴 실행에 사용되는 스레드를 확인하고 코루틴을 특정 스레드에서 실행하도록 지시할 수 있다.

코루틴 라이브러리의 withContext() 함수를 사용하여 코루틴 실행을 다른 스레드로 이동할 수 있다. 이를 코루틴 실행을 디스패치(dispatch)한다고 한다.

코루틴 코드를 기본 스레드 외부에서 실행하려면 기본 디스패처 또는 IO 디스패처에서 작업을 실행하도록 코루틴에 지시한다. 코루틴은 중지될 수 있으며 디스패처는 중지된 코루틴 실행을 재개한다. 스레드풀을 사용하는 디스패처를 통해 코루틴을 실행한다 해도 처음부터 끝까지 동일한 스레드에서 실행된다는 보장은 없다.

CoroutineDispatcher는 블로킹 IO 작업을 공유 스레드풀로 오프로드하도록 설계되었다. 즉, 블로킹 IO 작업 시 공유 스레드풀의 스레드가 추가로 생성되고 요청 시 종료된다. 디스패처의 작업에서 사용하는 스레드 수는 속성 값으로 제한할 수 있다.

launch, async를 포함한 모든 코루틴 빌더를 사용하여 코루틴 생성 시 CoroutineContext 선택 인자를 전달함으로써 디스패처를 명시할 수 있다. 코루틴 빌더는 코루틴을 생성하고 명시한 디스패처에 함수 본문의 실행을 전달한다. 컨텍스트를 인자로 전달하지 않을 경우 실행 중인 코루틴 스코프(상위 코루틴의 스코프)에서 컨텍스트를 상속받는다.

코틀린 표준 라이브러리에는 다음 세 종류의 디스패처가 존재한다. 세 가지 디스패처를 사용하여 코루틴을 실행할 스레드를 지정할 수 있다.

  • Dispatchers.Default: 기본 디스패처
  • Dispatchers.IO: IO 디스패처
  • Dispatchers.Unconfined: 무제한 디스패처


기본 디스패처인 Dispatchers.Default는 일반적인 풀인 공유 백그라운드 스레드풀을 사용한다. CPU를 많이 사용하는 작업을 기본 스레드 외부에서 실행하는데 최적화되어 있다. 따라서 기본 디스패처는 리스트 정렬, JSON 파싱 등 CPU 리소스를 소비하는 계산(compute) 집약적인 코루틴에 적합하다.

IO 디스패처인 Dispatchers.IO는 IO 작업용으로 예약된 스레드에서 독립적으로 코루틴이 실행되도록 한다. IO 작업을 기본 스레드 외부에서 실행하는데 최적화되어 있다. 파일 및 디스크 IO, 네트워크 IO 등 IO 집약적인 블록킹 작업을 위해 디자인된 주문형(on-demand) 공유 스레드풀(필요시 스레드가 생성됨)을 사용한다.

무제한 디스패처인 Dispatchers.Unconfined는 일반적으로 사용해서는 안 된다.


모든 것을 비동기로

스레드를 블로킹하는 작업을 코루틴으로 비동기 처리한다고 해도 상위 호출이 동기 호출이면 여전히 스레드는 블로킹되는 시점이 존재한다.

기본 스레드에서 동기 함수를 호출하면 호출 스레드가 블로킹된다. 이 함수는 기본 안전 함수가 아니다.


스레드 로컬


코루틴 취소

백그라운드에서 실행되는 코루틴에 대한 제어가 필요할 경우 코루틴 취소 기능을 사용할 수 있다. launch() 함수는 코루틴 실행 후 결과를 Job 객체로 반환하며 Job 객체를 사용하여 코루틴의 상태를 추적 및 관리할 수 있다. 이 Job 객체를 사용하여 실행 중인 코루틴을 취소하거나 코루틴이 완료된 이후에 나머지 코드를 실행하기 위해 코루틴 실행 완료를 기다리게 만들 수도 있다.

Job 객체의 확장 함수는 다음과 같다.

  • cancel(): 코루틴을 취소한다. cancel 함수를 호출하면 코루틴이 취소되고 이후 코루틴 코드는 실행되지 않는다.
  • join(): 코루틴이 완료될 때까지 코루틴을 중지한다. 코루틴 실행이 완료될 때까지 기다리게 한다.
  • cancelAndJoin(): 코루틴을 취소하고 코루틴이 완료될 때까지 중지한다.


코루틴 실행이 오래 걸리는 이유로 코루틴을 취소하고자 한다면 withTimeout() 함수를 사용하면 된다. withTimeout() 함수는 코루틴 안의 중단 함수 블록의 코드를 실행하고, 설정한 타임아웃을 초과하면 TimeoutCancellationException 예외를 던진다. 따라서 타임아웃으로 인해 발생하는 해당 예외를 캐치하여 예외 처리를 할 수 있다. 아니면 타임아웃 시에 예외를 던지는 대신 null을 리턴하는 withTimeoutOrNull을 사용할 수도 있다.


코루틴 예외 처리

중단 함수에서 예외를 발생시키는 경우 코루틴 내에서 해당 중단 함수가 발생시키는 예외를 try/catch 구문으로 폴백(fallback) 처리할 수 있다.

코루틴이 실행되면 코루틴 스코프 내의 모든 코루틴(모든 자식 코루틴)이 완료될 때까지 기다리고, 만약 코루틴이 하나라도 실패하면 나머지 코루틴을 취소한다. 따라서 코루틴 중 하나에서 예외가 발생하면 나머지 코루틴들의 실행도 취소된다. 아직 실행되지 않은 동기적 코루틴 실행 또는 이미 실행 중인 비동기적 코루틴 실행 모두 취소된다.

만약 하나의 코루틴이 실행 중 실패하더라도 나머지 코루틴을 계속 실행하고자 한다면 SupervisorJob을 사용한다. 코루틴 스코프 인스턴스 생성 시 Job이 아닌 SupervisorJob을 사용하면 자식 코루틴 중 하나가 실패하더라도 부모 코루틴과 나머지 자식 코루틴은 실패하지 않고 계속 실행될 수 있도록 만들 수 있다. 즉, 자식 코루틴들의 실행이 각각 독립적으로 일어난다.

자식 코루틴의 실패 또는 취소는 SupervisorJob의 실패를 일으키지 않으며 SupervisorJob의 자식 코루틴에도 영향을 주지 않는다. 따라서 SupervisorJob 자식 코루틴에 대해서 발생 가능한 예외를 별도의 로직으로 사용자화하여 처리하고자 하는 경우 SupervisorJob을 사용한다.


플로우

코루틴에서 플로우(Flow)는 단일 값만 반환할 수 있는 중지 함수와 달리 여러 값을 순차적으로(sequentially) 방출(emit)할 수 있는 타입이다. 중단 함수는 단일 값을 비동기적으로 계산(요청 스레드는 블로킹되지 않으며 다른 작업을 수행할 수 있다)하여 반환하는 반면 플로우는 여러 값을 비동기적으로 계산하여 반환한다.

중단 함수가 컬렉션이나 시퀀스를 반환하는 경우 중단 함수 블록 내에서 요소들에 대한 반복 작업을 스레드 논블로킹 방식으로 수행할 수 있지만 이러한 반복 작업은 동기적이므로 전체적으로 작업 처리에 소요되는 시간과 요소의 수를 곱한 만큼의 시간이 소요되며 중단 함수는 모든 요소들에 대한 작업이 완료된 후 결과를 한 번에 반환한다. 플로우를 사용할 경우 중단 함수를 사용하여 스레드를 블로킹하지 않는 것은 동일하지만 비동기적으로(동시적으로) 일련의 값을 처리한다.

컬렉션 및 시퀀스와 플로우의 차이점과 공통점은 다음과 같다.

  • 처리 방식
    • 컬렉션 및 시퀀스: 동기적으로 요소가 처리된다. 각 요소는 이전 요소의 처리가 완료된 후에만 처리된다.
    • 플로우: 비동기적으로 요소가 처리된다. 한 번에 여러 요소들이 동시적으로 처리될 수 있다.
  • 처리 순서
    • 컬렉션 및 시퀀스: 한 번에 하나의 요소가 순차적으로 처리되기 때문에 요소의 순서가 항상 보장된다.
    • 플로우: 비동기적으로 요소가 처리되지만 요소를 방출된 순서대로 수집한다. 요소는 생산자에 의해 생성된 순서대로 소비자에게 전달되므로 요소의 순서가 항상 보장된다.


플로우는 비동기, 논블로킹 방식으로 데이터에 대한 처리를 수행할 수 있는 데이터 스트림(data stream)의 개념이다. 코틀린 언어를 사용하여 리액티브 프로그래밍(reactive programming) 패러다임을 따르는데 사용되는 타입이기도 하다. 스프링 웹플럭스에서 사용하는 리액터 타입인 Flux와 개념적으로 유사하다. 플로우가 방출하는 값은 동일한 타입이어야 한다.

데이터 스트림은 크게 세 가지 엔티티와 관련이 있다.

  • 생산자 (producer): 스트림에 추가되는 데이터를 생성한다. 코루틴을 통해 플로우는 비동기적으로 데이터를 생성할 수 있다.
  • 중개자 (intermediates): 스트림으로 방출되는 각 값 또는 스트림 자체를 변경한다. 생산자와 소비자 사이에 위치하여 데이터 스트림을 수정하여 요구 사항에 맞게 조정하는 중개자 역할을 한다. 데이터 스트림 처리에서 중개자는 선택 사항이다.
  • 소비자 (consumer): 스트림의 값을 소비한다.


소비자의 예로는 모바일 네이티브 애플리케이션에서 사용자 인터페이스를 통해 최종적으로 데이터를 표시하고 사용자의 입력 이벤트를 처리하는 UI 레이어가 있으며, 생산자의 예로는 데이터베이스로부터 조회한 데이터를 생산하는 데이터 접근 레이어가 있다.

방출은 생산자가 소비자에게 데이터를 전달하는 것을 의미한다. 생산자는 데이터를 방출하고 소비자는 방출된 데이터를 수집한다. 방출된 데이터가 처리되는 과정은 비동기 및 동시적으로 일어나게 되지만 데이터는 생산자가 생성한 순서대로 수집된다. 이때, 소비자가 데이터를 수집하는 속도에 따라 생상잔자의 데이터 방출 속도가 조절되며 이를 리액티브 프로그래밍의 역압(backpressrue) 메커니즘이라고 한다.

플로우는 플로우 빌더 API를 사용하여 생성한다. 플로우 빌더 블록은 중단 가능하다. 플로우 빌더 블록 내에서 emit() 함수를 사용하여 플로우로부터 값을 방출하며, collect() 함수를 사용하여 플로우로부터 방출된 값을 수집한다.

// 플로우 빌더
flow {
  for (i in 1..3) {
    // 값 방출
    emit(i)
  }
}
// 
.collect { ... }


.asFlow() 확장 함수를 사용하여 다양한 컬렉션 및 시퀀스를 플로우로 변환할 수도 있다.

(1..3).asFlow().collect { ... }


플로우는 일련의 데이터들을 대상으로 다양한 연산을 수행하는 연산자들을 제공한다. 연산자에는 스트림 처리와 유사하게 중간 연산자과 종단 연산자가 있다.

  • 중간 연산자 (intermediate operator): 업스트림 플로우에 연산을 적용한 후 다운스트림 플로우로 반환한다. 기본적인 연산자에는 map(), filter() 등이 있다. 중간 연산자에 대한 호출은 중단 함수 자체가 아니므로 빠르게 작동하며 변환된 새로운 플로우의 정의를 반환한다. 플로우의 데이터를 변환 또는 필터링하거나 새로운 플로우를 만들기 위해 중간 연산을 적용한다.
  • 종단 연산자 (terminal operator): 플로우의 수집(collection)을 시작하는 중단 함수이다. 기본적인 연산자는 수집 연산자인 collect()이다. 종단 연산자는 플로우의 데이터를 최종적으로 수집하여 다른 컬렉션 데이터 타입(리스트나 셋)으로 변환하거나 플로우를 값으로 리듀싱(reducing)한다. 또한 중첩된 플로우를 평탄화(flattening)하는 연산이 수집 연산 전에 일어날 수도 있다.


컬렉션이나 시퀀스 대상의 연산자들과 다른 점은 연산자 내부의 코드 블록이 중단 함수를 호출할 수 있다는 것이다. 연산이 중단 함수에 의해 구현되는 장기 실행 작업인 경우에도 플로우에 유입되는 요청은 스레드를 블로킹하지 않으며 순차적으로 처리된다.

다중 플로우에서 사용되는 특수 연산자를 사용하지 않는 한 플로우의 개별적인 수집은 순차적으로 이루어진다. 수집은 종단 연산자를 호출하는 코루틴에서 수행되며 기본적으로 새로운 코루틴은 시작되지 않는다. 각각의 방출된 값은 업스트림에서 다운스트림으로 전달되는 과정에서 중간 연산자에 의해 처리된 후 종단 연산자에게 전달된다.

플로우 빌더는 코루틴 내에서 실행된다. 따라서 비동기 처리의 이점을 누릴 수 있지만 다음과 같이 몇 가지 제한 사항이 적용된다.

  • 플로우는 순차적이다. 생산자가 코루틴 내에 있으므로 중단 함수를 호출하면 중단 함수가 반환될 때까지 생산자가 중단된다. 예를 들어 네트워크를 통해 외부 API로부터 데이터를 조회하는 생산자는 네트워크 요청이 완료될 때까지 중단되며 네트워크 요청이 완료되면 조회 결과값은 스트림에 방출된다. 생산자가 중단되더라도 현재 스레드는 블로킹되지 않는다.
  • 기본적으로 플로우 빌더의 생산자는 종단 연산자를 통해 데이터를 수집하는 코루틴의 코루틴 컨텍스트에서 실행되며 다른 코루틴 컨텍스트에서 값을 방출할 수 없다. 따라서 새로운 코루틴을 생성하거나 withContext() 블록을 사용하여 다른 코루틴 컨텍스트에서 emit()을 호출하면 안 된다. 이 경우 callbackFlow()와 같은 다른 플로우 빌더를 사용할 수 있다. 플로우의 코루틴 실행 컨텍스트를 변경하려면 중간 연산자인 flowOn()를 사용한다.


플로우는 콜드 스트림(cold stream)이라는 특징을 가지고 있으므로 구독자(subscriber)가 구독(subscription)을 시작하기 전까지 생산자는 데이터를 생성하지 않는다. 이는 리소스를 효율적으로 사용할 수 있게 해준다.


플로우 평탄화

플로우는 중첩될 수 있다. 중첩된 플로우란 각 요소가 다시 플로우를 생성하는 형태를 말한다. 이러한 중첩된 구조는 비동기적인 코드에서 자주 발생한다. 예를 들어 네트워크 요청 결과로 받은 스트림 데이터를 처리할 때 스트림 데이터의 각 요소는 다시 비동기적인 작업을 수행할 수 있다. 플로우는 비동기적으로 수신된 값의 시퀀스를 나타내며 시퀀스의 각 값이 다른 값의 시퀀스에 대한 요청을 트리거하는 상황이 쉽게 발생할 수 있다.

정수를 인자로 전달 받은 후 값을 방출함으로써 문자열 플로우를 반환하는 함수를 정의하고 세 개의 정수로 구성된 플로우의 각각의 데이터에 대해 함수를 호출하는 예는 다음과 같다.

// 내부 플로우 처리
fun requestFlow(i: Int): Flow<String> = flow {
  emit("$i: First")
  delay(500)
  emit("$i: Second")
}

...

// 외부 플로우 처리
(1..3).asFlow().map {
  requestFlow(it)
}


위 경우 결과값은 플로우의 플로우 타입인 Flow<Flow<String>>이며 데이터에 대해 추가 처리를 위해서는 단일 플로우로 평탄화해야 한다. 컬렉션과 시퀀스에는 이를 위한 flatten() 및 flatMap() 연산자가 있다. 그러나 플로우의 경우 비동기적 특성으로 인해 다른 평탄화 처리가 필요하다. 플로우의 평탄화 연산자는 다음과 같다.

  • flattenConcat(), flatMapConcat(): 플로우의 플로우를 연결(concatenation)한다. 외부 플로우를 순차적 및 동기적으로 수집하기 시작하면 내부 플로우와 외부 플로우는 순차적 및 동기적으로 처리된다. 내부 플로우에 블로킹 작업이 존재하는 경우 작업 처리 시간은 블로킹 작업 시간을 포함한다. flatMapConcat() 함수 호출은 순차적으로 map { ... }을 먼저 수행한 다음 결과에 대해 flattenConcat() 함수를 호출하는 것과 동일하다.
  • flattenMerge(), flatMapMerge(): 플로우의 플로우를 병합(merge)한다. 외부 플로우를 동시적으로 수집하며 그 값을 하나의 플로우로 병합하여 가능한 한 빨리 값이 방출되도록 한다. concurrency 파라미터를 사용하여 동시에 수집되는 동시 플로우 수를 제한할 수 있다. flatMapMerge() 함수 호출은 순차적으로 map { ... }을 먼저 수행한 다음 결과에 대해 flattenMerge() 함수를 호출하는 것과 동일하다.


외부 플로우의 데이터에 대해 블로킹 작업을 수행한 후 내부 플로우의 데이터에 대해 다른 블로킹 작업을 수행하고, 외부 플로우의 블로킹 작업 소요 시간이 내부 플로우의 블로킹 작업 보다 짧은 경우 플로우 연결의 예는 다음과 같다. flatMapConcat() 함수는 결과 플로우인 외부 플로우를 순차적으로 수집하며 내부 플로우의 코드 블록을 순차적으로 호출함으로써 데이터를 순차적으로 방출한다. 외부 플로우의 데이터 하나에 대해 내부 플로우의 모든 방출 작업이 순차적으로 완료된 후 외부 플로우의 그 다음 데이터에 대한 처리가 수행된다.

fun requestFlow(i: Int): Flow<String> = flow {
  emit("$i: First")
  delay(500)
  emit("$i: Second")
}

...

val startTime = System.currentTimeMillis()
(1..3).asFlow().onEach { delay(100) }
    .flatMapConcat { requestFlow(it) }                                                                           
    .collect { value ->
        println("$value at ${System.currentTimeMillis() - startTime} ms from start") 
    }

1: First at 121 ms from start
1: Second at 622 ms from start
2: First at 727 ms from start
2: Second at 1227 ms from start
3: First at 1328 ms from start
3: Second at 1829 ms from start


외부 플로우의 블로킹 작업 소요 시간이 내부 플로우의 블로킹 작업 보다 긴 경우에도 순차적 및 동기적으로 처리된다.

val startTime = System.currentTimeMillis()
(1..3).asFlow().onEach { delay(1000) }
    .flatMapConcat { requestFlow(it) }                                                                           
    .collect { value ->
        println("$value at ${System.currentTimeMillis() - startTime} ms from start") 
    }

1: First at 1048 ms from start
1: Second at 1549 ms from start
2: First at 2549 ms from start
2: Second at 3049 ms from start
3: First at 4050 ms from start
3: Second at 4550 ms from start


외부 플로우의 블로킹 작업 소요 시간이 내부 플로우의 블로킹 작업 보다 짧은 경우 플로우 병합의 예는 다음과 같다. 내부 블로우의 블로킹 작업 시간 내에 동시적으로 처리할 수 있는 작업을 외부 플로우의 모든 데이터를 대상으로 먼저 처리한다. flatMapMerge() 함수는 결과 플로우인 외부 플로우를 동시적으로 수집하며 코드 블록을 순차적으로 호출함으로써 내부 플로우의 데이터를 순차적으로 방출한다.

val startTime = System.currentTimeMillis()
(1..3).asFlow().onEach { delay(100) }
    .flatMapMerge { requestFlow(it) }                                                                           
    .collect { value ->
        println("$value at ${System.currentTimeMillis() - startTime} ms from start") 
    }

1: First at 136 ms from start
2: First at 231 ms from start
3: First at 333 ms from start
1: Second at 639 ms from start
2: Second at 732 ms from start
3: Second at 833 ms from start


그러나 내부 플로우의 블로킹 작업 시간 내에 동시적으로 처리할 수 없는 데이터에 대한 작업의 경우 동시적으로 처리되지 않을 수 있다.

val startTime = System.currentTimeMillis()
(1..10).asFlow().onEach { delay(100) }
    .flatMapMerge { requestFlow(it) }                                                                           
    .collect { value ->
        println("$value at ${System.currentTimeMillis() - startTime} ms from start") 
    }

1: First at 253 ms from start
2: First at 345 ms from start
3: First at 446 ms from start
4: First at 546 ms from start
5: First at 647 ms from start
6: First at 748 ms from start
1: Second at 754 ms from start
2: Second at 846 ms from start
7: First at 848 ms from start
3: Second at 946 ms from start
8: First at 949 ms from start
4: Second at 1047 ms from start
9: First at 1049 ms from start
5: Second at 1148 ms from start
10: First at 1150 ms from start
6: Second at 1250 ms from start
7: Second at 1349 ms from start
8: Second at 1449 ms from start
9: Second at 1550 ms from start
10: Second at 1651 ms from start


외부 플로우의 블로킹 작업 소요 시간이 내부 플로우의 블로킹 작업 보다 길다면 작업 시간 면에서 비동기 및 동시적 처리의 이점은 없으며 순차적 및 동기적으로 처리된다.

val startTime = System.currentTimeMillis()
(1..3).asFlow().onEach { delay(1000) }
    .flatMapMerge { requestFlow(it) }                                                                           
    .collect { value ->
        println("$value at ${System.currentTimeMillis() - startTime} ms from start") 
    }

1: First at 1145 ms from start
1: Second at 1646 ms from start
2: First at 2145 ms from start
2: Second at 2646 ms from start
3: First at 3146 ms from start
3: Second at 3646 ms from start


외부 플로우의 데이터에 대해 블로킹 작업을 수행하지 않는 경우 플로우 연결의 예는 다음과 같다. 외부 플로우의 데이터에 대한 내부 플로우의 작업이 모두 완료되어야 다음 데이터에 대한 처리가 이루어지므로 모든 작업은 순차적 및 동기적으로 처리된다.

val startTime = System.currentTimeMillis()
(1..3).asFlow()
    .flatMapConcat { requestFlow(it) }                                                                           
    .collect { value ->
        println("$value at ${System.currentTimeMillis() - startTime} ms from start") 
    }

1: First at 32 ms from start
1: Second at 556 ms from start
2: First at 556 ms from start
2: Second at 1056 ms from start
3: First at 1057 ms from start
3: Second at 1557 ms from start


외부 플로우의 데이터에 대해 블로킹 작업을 수행하지 않는 경우 플로우 병합의 예는 다음과 같다.

val startTime = System.currentTimeMillis()
(1..3).asFlow()
    .flatMapMerge { requestFlow(it) }                                                                           
    .collect { value ->
        println("$value at ${System.currentTimeMillis() - startTime} ms from start") 
    }

1: First at 133 ms from start
3: First at 134 ms from start
2: First at 146 ms from start
3: Second at 636 ms from start
1: Second at 646 ms from start
2: Second at 647 ms from start


리스트 내 데이터에 대해 일괄 비동기 동시 처리

코틀린에서 리스트 내 데이터들에 대한 작업을 비동기 및 동시적으로(concurrently) 처리하기 위해 다음 방법 중 하나를 사용할 수 있다.

  • 코루틴 빌더 사용
  • 플로우(Flow) 사용


코루틴을 사용할 경우 리스트 내 데이터를 대상으로 launch 또는 async 코루틴 빌더를 사용하여 반복적으로 코루틴을 실행한다. 작업 결과가 필요하지 않다면 launch 빌더를 사용하고, 반복적인 작업에 대한 결과를 생성하려면 async 빌더를 사용하여 코루틴을 실행한 후 awaitAll() 함수를 호출하여 모든 작업들이 완료되기를 기다린 다음 작업 결과를 반환한다.

listOf(1, 2, 3).forEach {
  launch {
    비동기 처리 작업
  }
}

val deferredList = listOf(1, 2, 3).map {
  async {
    비동기 처리 작업
  }
}

val results = deferredList.awaitAll()


코루틴 대신 비동기 스트림 처리를 위한 플로우 타입을 사용할 수도 있다. 플로우를 사용하여 데이터 스트림을 통해 여러 데이터를 비동기 및 동시적으로 처리할 수 있다. 리스트를 플로우로 변환하고 플로우 빌더를 사용하여 비동기 처리 작업을 수행한다. 이때 flatMapMerge() 연산자를 사용한다.

listOf(1, 2, 3).asFlow()
  .flatMapMerge(concurrency = 3) { value ->
    flow {
      비동기 처리 작업
      emit(결과)
    }
  }.collect { result ->
    ...
  }


리액터와 코루틴

코틀린은 리액티브 스트림 구현체인 리액터(Reactor)와 코루틴 간의 통합을 위해 별도의 패키지를 통해 여러 타입과 함수를 제공한다. 리액터의 컨텍스트를 코루틴 컨텍스트로 래핑, 코루틴의 결과값을 리액터 타입으로 변환, 리액터 타입(모노(Mono) 또는 플럭스(Flux))으로부터 값을 대기(await)하여 결과값을 조회하는 등의 기능을 제공한다. 주요 함수는 다음과 같다.

  • ContextView.asCoroutineContext(): 주어진 ContextView를 ReactorContext로 래핑하여 코루틴의 컨텍스트에 추가하고 나중에 ReactorContext를 통해 사용할 수 있도록 한다. asCoroutineContext()는 ReactorContext 요소를 CoroutineContext에 넣는다. CoroutineContext는 코루틴을 통해 리액터의 컨텍스트에 대한 정보를 전파하는데 사용된다.
  • Flow.asFlux(): 주어진 플로우를 콜드(cold) 플럭스 타입으로 변환한다. 플럭스 구독자가 폐기되면 원래 플로우는 취소된다.
  • Deferred.asMono(): 지연된 값(Deferred)을 성공 또는 에러를 알리는 핫(hot) 모노 타입으로 변환한다.
  • Job.asMono(): 해당 잡이 완료되면 성공 신호를 보내는 핫 모노 타입으로 변환한다.
  • Mono.awaitSingle(): 스레드를 블로킹하지 않고 주어진 모노에서 단일값을 기다렸다가 결과값을 반환하거나, 모노가 에러를 생성한 경우 또는 값이 없는 경우 해당 예외를 던진다. 모노가 값 없이 완료되더라도 null이 반환되지 않는다.
  • Mono.awaitSingleOrNull(): 스레드를 차단하지 않고 주어진 모노에서 단일값을 기다렸다가 결과값을 반환하거나, 발행자가 에러를 생성한 경우 해당 예외를 던진다. 모노가 값 없이 완료되면 null이 반환된다.
  • mono(): 코루틴에서 주어진 블록을 실행하고 그 결과를 방출하는 콜드 모노를 생성한다. 반환된 모노가 구독될 때마다 새 코루틴이 시작된다. 코루틴 스코프 블록 실행 결과가 null인 경우 MonoSink.success()가 값 없이 호출된다. 구독을 취소하면 실행 중인 코루틴이 취소된다.


코루틴 디버깅

코루틴 라이브러리는 비동기 프로그램의 디버깅을 좀더 쉽게 도와주는 기능을 제공한다.

kotlinx.coroutines.debug 시스템 속성을 사용하여 JVM에서 코루틴을 디버그 모드로 실행할 수 있다. 즉, 프로그램을 -Dkotlinx.coroutines.debug 플래그와 함께 실행한다. 디버그 모드는 실행된 모든 코루틴에 고유한 이름을 부여한다. 코루틴 스코프 코드 내에서 Thread.currentThread().name를 사용하여 현재 코드를 실행 중인 코루틴의 이름을 디버깅에 사용할 수 있다.


코루틴 코드 작성

  • ()는 비동기 동시 실행

  • runBlocking 
    fun main() {
    메인 코드1
    // 메인 스레드가 블로킹된다. 코루틴 실행이 완료될 때까지 이후 메인 스레드 작업이 수행되지 않는다.
    runBlocking { // 코루틴 스코프 생성
        // 코루틴이 실행된다. 중단 함수이므로 코루틴이 실행되다가 중지될 수 있다. 코루틴이 중지된 경우 중지된 코루틴이 재개된다.
        코루틴 코드 (중단 함수)
    }
    // 코루틴 실행이 완료되면 블로킹되었던 메인 스레드의 나머지 작업이 다시 수행된다.
    메인 코드2
}
    • 실행 순서: 메인 코드1 실행 -> 코루틴 코드 실행 -> 메인 코드2 실행
  • launch 
    fun main() {
    메인 코드1
    // 메인 스레드가 블로킹된다. 코루틴 실행이 완료될 때까지 이후 메인 스레드 작업이 수행되지 않는다.
    runBlocking { // 코루틴 스코프 생성
        // 코루틴이 실행된다. 중단 함수이므로 코루틴이 실행되다가 중지될 수 있다. 코루틴이 중지된 경우 중지된 코루틴이 재개된다.
        외부 코루틴 코드1 (중단 함수)
        launch { // 코루틴 스코프 확장
            // 새로운 내부 코루틴이 실행된다.
            내부 코루틴 코드 (중단 함수)
        }
        // 내부 코루틴이 실행 완료되지 않아도 외부 코루틴이 실행된다.
        외부 코루틴 코드2 (중단 함수)
    }
    // 모든 코루틴 실행이 완료되면 블로킹되었던 메인 스레드의 나머지 작업이 다시 수행된다.
    메인 코드2
}
    • 실행 순서: 메인 코드1 실행 -> 외부 코루틴 코드1 실행 -> (내부 코루틴 코드 실행, 외부 코루틴 코드2 실행) -> 메인 코드2 실행
    • 동기 실행
      • 함수 내 동기 실행: 메인 코드1 -> runBlocking 코루틴 코드 -> 메인 코드2
      • runBlocking 코루틴 코드 내 동기 실행: 외부 코루틴 코드1 -> (내부 코루틴 코드, 외부 코루틴 코드2)
    • runBlocking 코루틴 코드 내 비동기 실행: launch 내부 코루틴 코드, 외부 코루틴 코드2
  • async, await 
    fun main() {
    메인 코드1
    // 메인 스레드가 블로킹된다. 코루틴 실행이 완료될 때까지 이후 메인 스레드 작업이 수행되지 않는다.
    runBlocking { // 코루틴 스코프 생성
        // 코루틴이 실행된다. 중단 함수이므로 코루틴이 실행되다가 중지될 수 있다. 코루틴이 중지된 경우 중지된 코루틴이 재개된다.
        외부 코루틴 코드1 (중단 함수)
        val asyncCoroutine = async { // 코루틴 스코프 확장
            // 새로운 내부 코루틴이 실행된다.
            내부 코루틴 코드 (중단 함수)
        }
        // 내부 코루틴이 실행 완료되지 않아도 외부 코루틴이 실행된다.
        외부 코루틴 코드2 (중단 함수)
        // await 함수는 내부 코루틴이 실행 완료될 때까지 코루틴을 중지한다. 현재 코루틴 실행 스레드를 블로킹하지는 않으며 스레드는 다른 작업을 수행할 수 있다.
        asyncCoroutine.await()
        // await 함수로 인해 내부 코루틴이 실행 완료되면 나머지 외부 코루틴이 실행된다. 
        외부 코루틴 코드3 (중단 함수)
    }
    // 모든 코루틴 실행이 완료되면 블로킹되었던 메인 스레드의 나머지 작업이 다시 수행된다.
    메인 코드2
}
    • 실행 순서: 메인1 실행 -> 외부 코루틴1 실행 -> (내부 코루틴 실행, 외부 코루틴2 실행) -> 내부 코루틴 완료 -> 외부 코루틴3 실행 -> 메인2 실행
    • 동기 실행
      • 함수 내 동기 실행: 메인 코드1 -> runBlocking 코루틴 코드 -> 메인 코드2
      • runBlocking 코루틴 코드 내 동기 실행: 외부 코루틴 코드1 -> (내부 코루틴 코드, 외부 코루틴 코드2) -> 외부 코루틴 코드3
    • runBlocking 코루틴 코드 내 비동기 실행: async 내부 코루틴 코드, 외부 코루틴 코드2
  • withContext 
    fun main() {
    메인 코드1
    // 메인 스레드가 블로킹된다. 코루틴 실행이 완료될 때까지 이후 메인 스레드 작업이 수행되지 않는다.
    runBlocking { // 코루틴 스코프 생성
        // 코루틴이 실행된다. 중단 함수이므로 코루틴이 실행되다가 중지될 수 있다. 코루틴이 중지된 경우 중지된 코루틴이 재개된다.
        외부 코루틴 코드1 (중단 함수)
        // withContext 함수는 내부 코루틴이 실행 완료될 때까지 코루틴을 중지한다. 현재 코루틴 실행 스레드를 블로킹하지는 않으며 스레드는 다른 작업을 수행할 수 있다.
        withContext { // 코루틴 스코프 확장
            // 새로운 내부 코루틴이 실행된다.
            내부 코루틴 코드1 (중단 함수)
        }
        withContext { // 코루틴 스코프 확장
            // 새로운 내부 코루틴이 실행된다.
            내부 코루틴 코드2 (중단 함수)
        }
        // withContext 함수로 인해 내부 코루틴이 실행 완료되면 나머지 외부 코루틴이 실행된다. 
        외부 코루틴 코드2 (중단 함수)
    }
    // 모든 코루틴 실행이 완료되면 블로킹되었던 메인 스레드의 나머지 작업이 다시 수행된다.
    메인 코드2
}
    • 실행 순서: 메인1 실행 -> 외부 코루틴1 실행 -> 내부 코루틴 실행 -> 내부 코루틴 완료 -> 외부 코루틴2 실행 -> 메인2 실행
    • 동기 실행
      • 함수 내 동기 실행: 메인 코드1 -> runBlocking 코루틴 코드 -> 메인 코드2
      • runBlocking 코루틴 코드 내 동기 실행: 외부 코루틴 코드1 -> 내부 코루틴 코드1 -> 내부 코루틴 코드2 -> 외부 코루틴 코드2
    • runBlocking 코루틴 코드 내 비동기 실행: 없음
  • Dispatchers 
    fun main() = runBlocking {
    외부 코루틴 코드1 (중단 함수)
    launchWithIO()
    launchWithDefault()
}

suspend fun launchWithIO() {
    withContext(Dispatchers.IO) {
        내부 코루틴 코드1 (중단 함수)
    }
}

suspend fun launchWithDefault() {
    withContext(Dispatchers.Default) {
        내부 코루틴 코드2 (중단 함수)
    }
}
    • 실행 순서: 외부 코루틴1 실행 -> 내부 코루틴1 실행 -> 내부 코루틴1 실행 완료 -> 내부 코루틴2 실행 -> 내부 코루틴2 실행 완료
    • 동기 실행: 외부 코루틴 코드1 -> 내부 코루틴 코드1 -> 내부 코루틴 코드2
    • runBlocking 코루틴 코드 내 비동기 실행: 없음
  • 코루틴 취소하기 
    fun main() = runBlocking {
    val job = launch {
      내부 코루틴 코드 (중단 함수) 
    }
    외부 코루틴 코드1 (중단 함수)
    job.cancel()
    job.join()
    외부 코루틴 코드2 (중단 함수)
}
    • 실행 순서: (내부 코루틴 실행, 외부 코루틴1 실행) -> 내부 코루틴 중지 -> 외부 코루틴2 실행


코루틴 코드 작성 예제

fun main() {
    // 메인 스레드
    println("Before creating coroutine")
    // 메인 스레드가 블로킹된다.
    runBlocking { // 코루틴 스코프 생성
        // 코루틴이 실행된다.
        print("Hello, ")
        // 논블로킹 지연에 의해 코루틴이 실행되다가 중지된다. 스레드는 블로킹되지 않는다.
        delay(200L)
        // 지연 후에 중지된 코루틴이 재개된다.
        println("World!")
    }
    // 코루틴 실행이 완료되면 블로킹되었던 메인 스레드의 나머지 작업이 다시 수행된다.
    println("After coroutine is finished")
}

Before creating coroutine
Hello, World!
After coroutine finished

fun main() = runBlocking { // 코루틴 스코프 생성
    launch { // 새로운 코루틴을 생성하고 실행한다.
        // 논블로킹 지연에 의해 코루틴이 실행되다가 중지된다. 스레드는 블로킹되지 않는다.
        delay(1000L)
        // 지연 후에 중지된 코루틴이 재개된다.
        println("World!")
    }
    // 내부 코루틴이 실행 완료되지 않아도 외부 코루틴이 실행된다.
    // 내부 코루틴이 지연되는 동안 메인 코루틴이 실행된다.
    println("Hello")
}

Hello
World!

fun main() {
    // 메인 스레드가 블로킹된다.
    runBlocking { // 코루틴 스코프 생성
        // 코루틴이 실행된다.
        launch { // 새로운 코루틴을 생성하고 실행한다.
            suspendingWork()
        }
        // 내부 코루틴이 실행 완료되지 않아도 외부 코루틴이 실행된다.
        // 내부 코루틴이 지연되는 동안 메인 코루틴이 실행된다.
        println("Hello")
    }
    // 블로킹되었던 메인 스레드의 나머지 작업이 다시 수행된다.
    println("World.")
}

suspend fun suspendingWork() {
    // 논블로킹 지연에 의해 코루틴이 실행되다가 중지된다. 스레드는 블로킹되지 않는다.
    delay(1000L)
    // 지연 후에 중지된 코루틴이 재개된다.
    println("The")
}

Hello
The
World!


코루틴과 자바의 동시성 프로그래밍 비교


로깅


코틀린 테스트

코틀린은 여러 테스트 라이브러리에 대한 퍼사드를 제공하는 kotlin.test 표준 라이브러리를 제공한다. kotlin.test 라이브러리는 사용하는 테스트 프레임워크와 독립적으로 테스트 함수와 단정을 수행하기 위한 여러 함수를 대상으로 적용할 수 있는 어노테이션을 제공한다. 테스트 프레임워크는 Asserter 클래스를 통해 추상화되며 기본 Asserter 구현이 제공된다. Asserter 클래스는 테스트에서 직접 사용하기 위한 것은 아니며, 대신 Asserter에 위임하는 최상위 단정 함수를 사용하면 된다. kotlin.test를 사용하기 위해서는 다음과 같이 의존성을 추가한다. 이 의존성은 kotlin.test와 JUnit을 같이 사용할 수 있게 해준다.

dependencies {
  testImplementation(kotlin("test"))
}

그래들의 테스트 태스크 추가를 위해서는 다음과 같이 설정한다.

tasks.test {
  useJUnitPlatform()
}

코틀린 플러그인은 자동으로 코틀린 테스트 관련 의존관계를 처리해준다. 자바의 단위 테스트 프레임워크인 JUnit 대신 코틀린 전용 단위 테스트 프레임워크인 Kotest를 사용할 수 있다.

코틀린 코드를 단위 테스트할 때 목 객체를 사용하는 경우 Mockito 또는 MockK 라이브러리를 사용할 수 있다.

MockK은 코틀린의 기능에 대한 1급 지원, 코루틴 지원을 제공하며 final 클래스와 메서드에 대한 목 생성을 지원한다.

Mockito를 사용할 경우 코루틴 지원이 되지 않으므로 코루틴에 대한 테스트 시 일반 함수인 runTest()를 사용해야 한다. 반면 MockK은 coEvery(), coVerify() 등과 같이 중단 함수를 인자로 전달 받아 실행하는 일반 함수를 제공하므로 이 메서드들을 사용함으로써 테스트 함수가 일반 함수만을 호출하도록 하면 runTest() 함수 호출은 필요 없어진다.


Mockito 라이브러리

  • 목 객체 생성
  • 스텁 생성 
    `when`(목_객체.테스트_대상_함수()).then { 예상_반환_결과 }


MockK 라이브러리


1. 목 객체 생성
mockk<목_객체>()
mockk<목_객체>(relaxed = true)
  • 목 객체 생성 시 relaxed 값을 true로 설정하여 해당 목 객체의 모든 함수가 단순한 값을 반환하도록 할 수 있다.


2. 목 객체 생성 (어노테이션 사용)
@MockK
lateinit var 목_객체

@RelaxedMockK
lateinit var 목_객체

@BeforeEach
fun init() {
    MocKAnnotations.init(this)
}


3. 스텁 생성
  • 일반 함수 
    every { 목_객체.테스트_대상_일반_함수() } returns 예상_반환_결과
every { 목_객체.테스트_대상_일반_함수() } answers { 예상_반환_결과 }
every { 목_객체.테스트_대상_일반_함수() } throws 예상_발생_예외
  • 코루틴 함수 
    coEvery { 목_객체.테스트_대상_중지_함수() } returns 예상_반환_결과
coEvery { 목_객체.테스트_대상_중지_함수() } coAnswers { 예상_반환_결과 }


4. 테스트 대상 함수 호출 검증
  • 일반 함수 
    verify { 목_객체.테스트_대상_일반_함수() }
    • verify() 함수는 목 객체의 테스트 대상 함수가 호출되었는지 여부를 확인하여 검증한다.
  • 코루틴 함수 
    coVerify { 목_객체.테스트_대상_중지_함수() }
  • 호출 순서 검증 
    verifySequence { 목_객체.테스트_대상_일반_함수() }
verifyOrder { 목_객체.테스트_대상_일반_함수() }
    • verifySequence: 모든 함수들의 호출 순서 검증
    • verifyOrder: 특정 함수들의 호출 순서 검증


코루틴 테스트

코루틴을 테스트하기 위해서는 별도의 라이브러리를 사용하는 것이 좋다.

테스트 코드에서 중단 함수를 실행하기 위해서는 코루틴을 정의해야 한다. JUnit 테스트 함수 자체는 중단 함수가 아니며 중단 함수로 정의하면 테스트 시 에러가 발생한다. 따라서 별도의 코드 작성 방법을 따라야 한다.


테스트 시 주의점

JUnit을 사용하여 테스트 함수를 runBlocking으로 정의한 경우 블록 내에서 값을 반환하지 않도록 주의한다. 값을 반환하는 경우 해당 값의 타입을 반환 타입으로 하는 테스트 함수가 선언되어 테스트 함수 실행 시 No test found for given includes... 에러가 발생할 수 있다. JUnit은 기본적으로 반환 타입이 없는 메서드를 테스트 메서드로 인식하여 실행한다.

@Test
fun test() = runBlocking {
  반환값
}


블록하운드

블록하운드(blockhound)는 스레드를 블로킹하는 호출을 감지하는 자바 에이전트이다. 블록하운드는 JVM 클래스를 계측하여 애플리케이션 실행 도중 블로킹 호출을 감지하고 런타임 에러를 발생시킨다.

BlockHound.install() 메서드를 단순히 호출하여 애플리케이션을 시작하는 경우 코루틴 기능 동작 시 코틀린 리플렉션 API에 의한 블로킹 호출이 감지될 수 있다. 이를 막으려면 코틀린이 제공하는 org.jetbrains.kotlinx:kotlinx-coroutines-debug 모듈을 프로젝트에 추가하면 된다. 이 모듈은 코루틴을 트래킹하는 디버그 JVM 에이전트를 제공하는 역할을 하며, 코루틴 컨텍스트에서 블로킹 작업이 호출된 시점을 감지하는 자동 블록하운드 통합(integration) 기능도 제공한다.

블록하운드를 사용하여 외부 코루틴을 실행하는 스레드를 대상으로 블로킹 호출을 감지할 수 있다. 내부 코루틴은 감지 대상이 아니다.

블락하운드가 검출하는 코루틴 컨텍스트 내 블로킹 호출 예는 다음과 같다.

withContext(Dispatcher.IO) {
  Thread.sleep(1000)
}

coroutineScope {
  Thread.sleep(1000)
}


블락하운드가 검출하지 못하는 코루틴 컨텍스트 내 블로킹 호출 예는 다음과 같다.

coroutineScope {
  launch {
    Thread.sleep(1000)
  }
}


자바와 코틀린 통합

자바 코드에서 코틀린의 suspend 함수 호출


스프링 프레임워크와 코루틴

빈 정의 메서드

구성 클래스의 빈 정의 메서드(@Bean 어노테이션이 설정된 메서드)에서 코루틴을 사용하기 위해 GlobalScope 또는 runBlocking 코루틴 빌더를 사용할 수 있다. 빈 정의 메서드를 중단 함수가 아닌 일반 함수로 선언하고 코루틴을 정의한다.

GlobalScope 빌더는 전역 스코프에서 코루틴을 실행한다. 특정 작업에 구속되지 않는 코루틴 스코프인 GlobalScope은 전체 애플리케이션 라이프사이클 동안 실행되며 조기에 취소되지 않는 최상위 코루틴을 실행하는데 사용된다. GlobalScope 빌더를 통한 코루틴은 애플리케이션의 생명주기와 독립적으로 실행되며 주로 백그라운드에서 장기 실행 작업을 처리할 때 사용된다. GlobalScope는 애플리케이션의 다른 부분과 생명주기가 연결되지 않기 때문에 메모리 누수의 위험이 있으므로 주의해서 사용해야 한다.

runBlocking 빌더는 현재 스레드를 블로킹하면서 코루틴을 실행한다. 주로 테스트 코드나 메인 함수에서 애플리케이션의 초기화 시 특정 작업을 수행할 때 사용된다. 애플리케이션 첫 구동 시에만 호출되는 빈 정의 메서드의 경우 현재 스레드를 블로킹하는 runBlocking 빌더를 사용하더라도 성능 문제를 일으킬 가능성은 낮다.


트레일링 람다

코틀린에서 함수는 1급 클래스이므로 함수의 파라미터로 선언할 수 있고 함수 호출 시 인자로 전달할 수도 있다. 익명 함수인 람다를 파라미터로 전달받는 함수는 고차 함수가 된다. 함수의 마지막 인자로 전달되는 람다를 트레일링 람다(trailing lambda)라고 한다.

함수의 마지막 파라미터가 함수인 경우 해당 인자로 전달할 람다식은 괄호 밖에 배치할 수 있다.

val product = items.fold(1) { acc, e -> acc * e }


해당 함수의 인자가 람다식 하나일 경우 괄호를 완전히 생략할 수 있다.

run { ... }


스프링 프레임워크 사용 시 코틀린의 함수를 대상으로 스프링의 AOP를 적용할 수 있다. 그러나 코루틴의 중단 함수의 경우 AOP 적용에 있어 일부 기능 지원이 완벽하지 않으며 스프링 프레임워크 6.1.0 버전 이전에서는 중단 함수에 대한 AOP를 적용하기 위해 추가적인 처리가 필요하다링크 이 경우 트레일링 람다 구문을 사용하면 함수의 파라미터인 람다식 전 후로 작업을 실행함으로써 AOP 기능을 간단하게 구현할 수 있다.

fun process(function: (String) -> Unit) {
  람다식 호출 전 작업 수행
  val result = function.invoke()
  람다식 호출 후 작업 수행
}

...

process { it ->
  ...
}


중단 함수에 대해서도 적용 가능하다.

suspend fun process(suspendFunction: suspend () -> Unit) {
  람다식 호출 전 작업 수행
  val result = suspendFunction.invoke()
  람다식 호출 후 작업 수행
}

...

process {
   ...
}


resilience4j

resilience4j의 2.1.0 버전 기준으로 지원 어노테이션 및 스프링의 AOP는 중단 함수와 함께 작동하지 않는다링크. 따라서 resilience4j-kotlin 모듈링크을 사용하여 중지 함수를 데코레이션해야 한다.

서킷 브레이커, 레이트 리미터, 재시도, 타임 리미터 각각의 기능에 대해 다음 두 확장 함수를 사용할 수 있다.

  • executeSuspendFunction(): 중단 함수를 실행한다. 서킷 브레이커 내에서 중단 함수를 직접 실행하고 결과를 반환한다.
  • decorateSuspendFunction(): 중단 함수를 데코레이션 한 후 해당 기능에 대한 로직이 적용된 함수를 반환한다.


서킷 브레이커 패턴에서 executeSuspendFunction() 확장 함수를 사용하여 폴백 메서드를 실행하는 예는 다음과 같다.

val circuitBreaker = CircuitBreaker.ofDefaults()

try {
  val result = circuitBreaker.executeSuspendFunction {
    함수 실행
  }
  return result
} catch (e: Exception) {
  폴백 함수 실행
}


레코드 예외로 등록한 예외가 발생하여 서킷이 오픈되면 서킷 브레이커는 해당 예외를 던지며 폴백 메서드가 실행된다.

decorateSuspendFunction() 확장 함수를 사용하여 폴백 메서드를 실행하는 예는 다음과 같다.

val circuitBreaker = CircuitBreaker.ofDefaults()

val function = circuitBreaker.decorateSuspendFunction {
  중단 함수 실행
}

...

try {
  val result = function()
  return result
} catch (e: Exception) {
  폴백 함수 실행
}


decorateSuspendFunction() 확장 함수를 사용하여 중단 함수 또는 플로우를 데코레이션하더라도 서킷 브레이커는 중단 지점(suspension point)을 추가하지는 않는다. 데코레이션된 플로우의 경우 수집을 시도할 때 서킷이 오픈 상태이면 CallNotPermittedException 예외가 발생한다. 코루틴 스코프가 정상적으로 또는 예외적으로 취소되면 획득한 권한이 해제된다. 그 이후에는 성공 또는 실패 이벤트가 기록되지 않는다.


코루틴 플레이그라운드

https://pl.kotl.in/9zva88r7S


참고

Comments

You May Also Enjoy