Kotlin Concurrency

• 코루틴
• 동시성 통신
• 자바 동시성 사용

13.1 Coroutine

kotlin에서 동시성 처리를 위한 메커니즘으로 명령형 스타일의 코드로 작성해도 컴파일러가 코드를 효율적인 비동기 계산으로 자동변환한다. 즉, 일시 중단 가능한 함수 개념을 가지고 있다.

자바의 동시성 처리 단점? 멀티 스레드로 동시성 처리가 가능하지만 블러킹 연산(sleep, join, wait) 스레드를 실행 재개시 문맥 전환 비용, 스레드마다 시스템 자원이 많이 소모 비동기 연산을 위한 다양한 라이브러리가 있지만 일반적인 명령형 흐름제어가 아니라 복잡도가 높아짐(callback)

13.1.1 코루틴과 일시 중단 함수

코루틴의 라이브러리를 뒷받침하는 기본 요소는 일시 중단 함수

• 코루틴의 핵심은 일시 중단 함수 개념
• suspend 변경자로 표기
• 여러 가지 코루틴 빌더(launch, async, runBlocking)

suspend fun kotlin() {
  println("Hello")
  delay(100L) // 일시중단 함수 (스레드를 블럭시키지 않고 다른 작업 수행)
  println("World")
}

일반 함수에서는 일시 중단 함수를 호출할 수 없다.

fun kotlin() {
  println("Hello")
  delay(100L) // error: delay is a suspend function
  println("World")
}

13.1.2 코루틴 빌더

코루틴 빌더들

• launch { } : Job을 return 하는 coroutine
• async { } : Deffered 를 이용하여 single value를 return 하는 coroutine. Deferred value에. await()를 호출하여 최종 결과를 얻을 수 있다. Deferred는 Job이라, cancel 할 수도 있다.
• runBlocking { } : current thread 를 block 시키는 coroutine

launch vs async 비교

함수결과 반환반환 타입

launch	X	Job
async	O	Deferd<T>

launch() 빌더는 코루틴을 시작하고, 코루틴을 실행 중인 작업의 상태를 추적하고 변경할 수 있는 Job 객체를 반환한다.

• 아래 프로그램의 결과를 보면 병렬적으로 실행된 것을 확인 가능
• 다만, 실행 순서가 항상 일정하지는 않음

import kotlinx.coroutines.GlobalScope
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import java.lang.System.currentTimeMillis
​
fun main() { // main이 suspend 함수가 아님에 유의
    val time = currentTimeMillis()
​
    GlobalScope.launch {
        delay(100) // thread 블럭 시키지 않고 코드 수행만 멈추는 특별한 suspending function.
        println("Task 1 finished in ${currentTimeMillis() - time} ms")
    }
​
    GlobalScope.launch {
        delay(100)  
        println("Task 2 finished in ${currentTimeMillis() - time} ms")
    }
    // 스레드를 블럭
    Thread.sleep(200) 
}
​
// result
Task 2 finished in 176 ms
Task 1 finished in 176 ms

async() 빌더는 결과를 Deferred 인스턴스로 돌려주고, await() 메서드를 통해 계산 결과에 접근할 수 있다.

suspend fun main() {
    val message = GlobalScope.async {
        delay(100)
        "abc"
    }
​
    val count = GlobalScope.async {
        delay(100)
        1 + 2
    }
​
    delay(200)
    // await() 실행되면 코루틴은 일시중단되고 결과가 반환되면 코루틴은 다시 재개되고 결과를 출력
    val result = message.await().repeat(count.await())   // await()는 일시중단 가능한 코루틴 내부에서만 사용 가능
    println(result)
}
​
// result 
abcabcabc

Launch()와 async() 빌더는 일시 중단 함수 내부에서 스레드 호출을 블록 시키지는 않지만, 백그라운드 스레드를 공유하는 풀을 통해 작업을 실행한다.

runBlocking() 빌더는 디폴트로 현재 스레드에서 코루틴을 만들고 코루틴이 완료될 때까지 현재 스레드의 실행을 블럭시킨다.

• 코루틴이 끝나면 일시 중단 람다의 결괏값이 된다.
• 취소되면 예외를 발생.
• 블러킹 호출과 넌블러킹 호출 사이의 다리 역할을 위해 사용
• 테스트나 메인 함수에서 최상위 빌더로만 사용해야 함

import kotlinx.coroutines.*
​
fun main() {
    // 공유풀 백그라운드 스레드를 새로 할당 받아 실행
    GlobalScope.launch {
        delay(100)
        println("Background task: ${Thread.currentThread().name}")
    }
    // 메인 스레드에서 실행되고 메인 스레드를 블럭 시킴
    runBlocking {
        println("Primary task: ${Thread.currentThread().name}")
        delay(200)
    }
}
​
// result
Primary task: main @coroutine#2
Background task: DefaultDispatcher-worker-1 @coroutine#1

13.1.3 코루틴 영역과 구조적 동시성

동시성 작업 사이의 부모 자식 관계로 인해 이런 실행 시간제한이 가능하다. 어떤 코루틴을 다른 코루틴의 문맥에서 실행하면 후자가 전자의 부모가 된다. 이경우 자식의 실행이 모두 끝나야 부모가 끝날 수 있도록 부모와 자식의 생명 주기가 연관된다. 이런 기능을 구조적 동시성(structured concurrency)이라고 한다.

import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
​
fun main() {
    runBlocking {  // 부모 코루틴
        println("Parent task started ${Thread.currentThread().name}")
​
        launch { // 자식 코루틴
            println("Task A started ${Thread.currentThread().name}")
            delay(200)
            println("Task A finished ${Thread.currentThread().name}")
        }
​
        launch { // 자식 코루틴
            println("Task B started ${Thread.currentThread().name}")
            delay(200)
            println("Task B finished ${Thread.currentThread().name}")
        }
​
        delay(100)
        println("Parent task finished ${Thread.currentThread().name}")
    }
    println("Shutting down... ${Thread.currentThread().name}")
}
​
​
// result
Parent task started main @coroutine#1
Task A started main @coroutine#2
Task B started main @coroutine#3
Parent task finished main @coroutine#1 // 지연이 100이기에 runBlocking() 호출의 일시 중단 람다 본문이 먼저 출력됨
Task A finished main @coroutine#2
Task B finished main @coroutine#3
Shutting down... main

최상위 코루틴이 시작하고 CoroutineScope 인스턴스 안에서 launch를 호출해 두 가지 자식 코루틴을 시작한다.

실행 순서 : 부모 runBlocking 코루틴 -> 각 자식 launch 코루틴 시작 -> 자식 200 delay -> 부모 runBlocking 코루틴 실행 -> 각 자식 launch의 코루틴 출력 -> 자식 코루틴이 종료될 때까지 기다린 후 runBlocking 실행 -> 메인 스레드 블록 해제 후 main 출력되고 종료

coroutineScope()와 runBlocking()의 가장 큰 차이는 coroutineScope()가 일시 중단 함수라 현재 스레드를 블럭시키지 않는다는 점이다.

import kotlinx.coroutines.*
​
fun main() {
    runBlocking {  // 부모 코루틴
        println("Custom scope start")
​
        coroutineScope {  // 아래 자식 코루틴 실행이 끝날 때까지 일시 중단 처리 됨
            launch {  // 자식 코루틴
                delay(100)
                println("Task 1 finished")
            }
​
            launch {  // 자식 코루틴
                delay(100)
                println("Task 2 finished")
            }
        }
​
        println("Custom scope end")
    }
}
​
// result
Custom scope start main @coroutine#1
Task 1 finished main @coroutine#2
Task 2 finished main @coroutine#3
Custom scope end main @coroutine#1

13.1.4 코루틴 문맥

코루틴은 항상 특정 문맥에서 실행되는데 코루틴을 감싸는 변수 영역의 CoroutineContext 프로퍼티를 통해 이 문맥에 접근할 수 있고, 키-값 쌍으로 이뤄진 불변 컬렉션이다.

GlobalScope.launch {
  // 현재 잡을 얻고 "Task is active: true" 를 출력
  println("Task is active: $coroutineContext[Job.Key]!!.isActive")
}

13.2 코루틴 흐름 제어와 잡 생명 주기

잡은 동시성 작업의 생명 주기를 표현하는 객체다. 잡을 사용하면 작업 상태를 추적하고 필요할 때 작업을 취소할 수 있다.

Active: 작업이 시작됐고 아직 완료나 취소로 끝나지 않은 상태로, 보통 디폴트 상태.

• CoroutineStart.DEFAULT는 디폴트 동작이며, 잡을 즉시 시작한다.
• CoroutineStart.LAZY는 잡을 자동으로 시작하지 말라는 뜻으로, 잡이 New 상태가 되고 시작을 기다리게 된다.

Completing: 코루틴이 일시 중단 람다 블록의 실행을 끝낸 잡의 상태(완료 중)

• 자식들의 완료를 기다리는 상태로 잡은 모든 자식이 완료될 때까지 이 상태를 유지

Completed: 모든 자식이 완료된 상태(완료 됨)

New 상태의 잡에 대해 start() or join() 메서드를 호출하면 잡이 시작되면서 활성화 상태가 된다.

import kotlinx.coroutines.*

fun main() {
    runBlocking {  // 부모 코루틴
        val job = launch(start = CoroutineStart.LAZY) {  // 자식 코루틴
            println("Job started")
        }

        delay(100)

        println("Preparing to start...")
        job.start() // start 메서드가 호출되면 자식 코루틴 시작..
    }
}

// result
Preparing to start...
Job started

자식 코루틴의 시작을 부모 코루틴이 메시지를 호출한 뒤로 미룬다.

잡의 join() 메서드를 사용하면 조인 대상 잡이 완료될 때까지 현재 코루틴을 일시 중단시킬 수 있다. 두 자식 메시지의 실행이 끝난 후에 루트 코루틴이 출력되도록 보장해 준다.

import kotlinx.coroutines.*

fun main() {
    runBlocking {
        val job = coroutineContext[Job.Key]!!
        val jobA = launch { println("This is task A") }
        val jobB = launch { println("This is task B") }

        jobA.join()
        jobB.join()
				
        println("${job.children.count()} children running")
    }
}

// result
This is task A
This is task B
0 children running  

현재 잡의 상태를 추적할 수 있는 프로퍼티들.

상태가 "완료됨, "취소됨"인 잡의 isCompleted가 true이기에, Cancelled와 Completed는 isCancelled로 구분할 수 있다.

13.2.1 취소

잡의 cancel() 메서드를 호출하면 잡을 취소할 수 있다. 취소에는 협력이 필요하다. 즉, 취소 가능한 코루틴이 스스로 취소가 요청됐는지 검사해서 적절히 반응해줘야 한다.

import kotlinx.coroutines.*

// 코루틴을 취소했지만 취소를 위해 협력되지 않기에 취소 불가함
suspend fun main() {
    val squarePrinter = GlobalScope.launch(Dispatchers.Default) {
        var i = 1
        while (true) {
            println(i++)
        }
    }

    delay(100) // 자식 잡이 어느 정도 실행될 시간을 준다
    squarePrinter.cancel()
}

// isActive 프로퍼티를 현재 잡의 활성화 상태
suspend fun main() {
    val squarePrinter = GlobalScope.launch(Dispatchers.Default) {
        var i = 1
        while (isActive) {  // 취소 가능한 형태의 루프
            println(i++)
        }
    }

    delay(100) 
    squarePrinter.cancel()  // 부모 코루틴이 취소되면 자동으로 모든 자식의 실행을 취소
}

isActive 프로퍼티는 CoroutineScope 객체가 자체적으로 지원하는 프로퍼티로, 코루틴의 취소 상태를 Boolean 타입으로 갖고 있다. 부모 코루틴이 cancel() 메서드를 호출하면 squarePrinter의 상태가 취소 중으로 바뀌고, 그다음 isActive 검사를 통해 루프를 종료시킬 수 있다.

import kotlinx.coroutines.*

fun main() = runBlocking {
    val job = launch {
        repeat(1000) { i ->
            println("${Thread.currentThread().name} : 코루틴아 달려라 $i ...")
            delay(300L)
        }
    }

    delay(1000L)
    println("${Thread.currentThread().name} : 코루틴아 이제 그만 쉬어")

    //job.cancel()  // Job 취소
    //job.join()    // Job 끝날 때 까지 대기
    job.cancelAndJoin()  // 취소하고 끝날 때 까지 대기
    println("${Thread.currentThread().name} : 코루틴 종료")
}

// result
main @coroutine#2 : 코루틴아 달려라 0 ...
main @coroutine#2 : 코루틴아 달려라 1 ...
main @coroutine#2 : 코루틴아 달려라 2 ...
main @coroutine#2 : 코루틴아 달려라 3 ...
main @coroutine#1 : 코루틴아 이제 그만 쉬어
main @coroutine#1 : 코루틴 종료

13.2.2 타임아웃

작업이 완료되기를 무작정 기다리 수 없어서 타임아웃을 설정이 필요할 경우

• withTimeout() : TimeoutCancellationException 반환
• withTimeoutOrNull() : 예외 대신 null 반환

import kotlinx.coroutines.*
import java.io.File

fun main() {
    runBlocking {
        val asyncData = async { File("data.txt").readText() }
        try {
            val text = withTimeout(50) { asyncData.await() }
            println("Data loaded: $text")
        } catch (e: Exception) {
            println("Timeout exceeded")
        }
    }
}

13.2.3 코루틴 디스패치하기

코루틴은 스레드와 무관하게 일시 중단 가능한 계산을 구현할 수 있게 해 주지만, 코루틴을 실행하려면 여전히 스레드와 연관시켜야 한다. 코루틴 라이브러리에는 특정 코루틴을 실행할 때 사용할 스레드를 제어하는 특별한 코루틴 디스패처 컴포넌트가 있다.

• 디스패처는 코루틴 문맥의 일부
• launch() or runBlocking() 등에서 지정할 수 있음

import kotlinx.coroutines.*

fun main() {
  runBlocking {
    // 전역 스레드 풀 디스패처를 사용해 코루틴을 실행한다
    launch(Dispatchers.Default) {
      println(Thread.currentThread().name) // DefaultDispatcher-worker-1
    }
    println(Thread.currentThread().name) // main
  }
}

코루틴 디스패처는 병렬 작업 사이에 스레드를 배분해 주는 자바 실행기(executor)와 비슷함.

스레드 개수를 지정하려면 asCoroutineDispatcher() 확장 함수를 사용하여 코루틴 문맥으로 전환할 수 있다.

import kotlinx.coroutines.*
import java.util.concurrent.ScheduledThreadPoolExecutor
import java.util.concurrent.atomic.AtomicInteger

fun main() {
    val id = AtomicInteger(0)

    val executor = ScheduledThreadPoolExecutor(5) { runnable ->
        Thread(
            runnable,
            "WorkerThread-${id.incrementAndGet()}"
        ).also { it.isDaemon = true }  // 코루틴 종료 후 자동으로 종료
    }

    executor.asCoroutineDispatcher().use { dispatcher ->
        runBlocking {
            for (i in 1..6) {
                launch(dispatcher) {
                    println(Thread.currentThread().name)
                    delay(1000)
                }
            }
        }
    }
}

// result
WorkerThread-1 @coroutine#2
WorkerThread-4 @coroutine#5
WorkerThread-2 @coroutine#3
WorkerThread-3 @coroutine#4
WorkerThread-5 @coroutine#6
WorkerThread-2 @coroutine#7

asCoroutineDispatcher()를 호출하면 ExecutorCoroutineDispatcher를 반환하고, Closeable 인스턴스도 구현하기에 시스템 자원을 해제하려면 close() or use() 함수로 해제가 가능하다.

코루틴 라이브러리에서 제공하는 디스패처

• Dispatchers.Default: shared common pool 기반으로 pool 크기는 디폴트로 사용 가능한 CPU 코어 수이거나 일반적인 background job을 수행
• Dispatchers.IO: shared pool 기반으로 파일을 읽고 쓰는 것처럼 잠재적으로 블러킹 될 수 있는 I/O 작업에 최적화.
• Dispatchers.Main: 사용자 입력이 처리되는 UI 스레드에서 사용.(android view..)

newFixedThreadPoolContext() or newSingleThreadPoolContext()를 사용하여 새로운 스레드를 생성하거나 사용하는 디스패처를 만들 수도 있다. 다만, 실험적인 기능이라 추후 대치될 예정.

디스패처를 명시적으로 지정하지 않으면 코루틴이 시작한 영역부터 자동으로 상속된다.

import kotlinx.coroutines.*

fun main() {
    runBlocking { // main
        println("Root: ${Thread.currentThread().name}")

        launch {  // main
            println("Nested, inherited: ${Thread.currentThread().name}")
        }
				// 코루틴 문맥이 실행될 스레드 지정
        launch(Dispatchers.Default) {  // DefaultDispatcher-worker-1
            println("Nested, explicit: ${Thread.currentThread().name}")
        }
    }
}

// result
Root: main @coroutine#1
Nested, inherited: main @coroutine#2
Nested, explicit: DefaultDispatcher-worker-1 @coroutine#3

부모 코루틴이 없으면 암시적으로 Dispatchers.Default로 디스패처를 가정한다. 다만 runBlocking() 빌더는 현재 스레드를 사용한다. 특정 스레드에서만 실행하고 싶을 경우 withContext() 함수를 사용하면 가능하다.

import kotlinx.coroutines.*

@Suppress("EXPERIMENTAL_API_USAGE")
fun main() {
    newSingleThreadContext("Worker").use { worker ->
        runBlocking {
            println(Thread.currentThread().name)   // main
            withContext(worker) {
                println(Thread.currentThread().name) // Worker
            }
            println(Thread.currentThread().name)   // main
        }
    }
}

중단 가능 루틴의 일부를 한 스레드에서만 실행하고 싶을 때 유용하다.

13.2.4 예외 처리

lunch() 빌더는 예외를 부모 코루틴에게 전달하고, CoroutineExceptionHandler에 의해 처리된다.

• 부모 코루틴이 (자식에게서 발생한 오류와) 똑같은 오류로 취소되고, 나머지 자식도 모두 취소됨
• 자식들이 모두 취소되고 나면 부모는 예외를 코루틴 트리의 윗부분으로 전달(루트 코루틴)

import kotlinx.coroutines.*

fun main() {
    // 커스텀 핸들러..
    runBlocking {  // 최상위 코루틴
        launch { // 첫번째 자식 코루틴
            throw Exception("Error in task A")
            println("Task A completed")
        }

        launch { // 두번째 자식 코루틴
            delay(1000)
            println("Task B completed")
        }

        println("Root")
    }
}
// result
Root
Exception in thread "main" java.lang.Exception: Error in task A

최상위 코루틴 start -> 첫 번째 코루틴 예외 -> 최상위 코루틴 취소 -> 최상위 하위 자식 코루틴 취소

최상위에서 지정된 핸들러가 없기 때문에 Thread.uncauhtExceptionHandler에 의한 디펄트 동작을 실행 한다.

CoroutineExceptionHandler는 현재 코루틴 문맥과 던져진 예외를 인자로 전달받는다.

fun handleException(context: CoroutineContext, exception: throwable)

val handler = CoroutineExceptionHandler{ _, exception -> // 인자 2개인 람다를 받는 함수 생성
	println("Caught $exception")
}

코루틴 예외 핸들러의 인스턴스가 예외를 처리하도록 지정하려면 코루틴 문맥에 인스턴스를 넣어야 한다.

import kotlinx.coroutines.*

suspend fun main() {
    val handler = CoroutineExceptionHandler{ _, exception ->
        println("Caught $exception")
    }
    // 전역 영역에서 실행 및 handler 인스턴스 선언
    GlobalScope.launch(handler) {
        launch {
            throw Exception("Error in task A")
            println("Task A completed")
        }

        launch {
            delay(1000)
            println("Task B completed")
        }

        println("Root")
    }.join()
}
// result
Root
Caught java.lang.Exception: Error in task A

CoroutineExceptionHandler는 전역 영역에서 실행된 코루틴에 대해서만 정의할 수 있고, CoroutineExceptionHandler가 정의된 코루틴의 자식에 대해서만 적용된다. 앞선 runBlocking()의 경우 코루틴이 전역 영역에서 실행되지 않아 디펄트 핸들러를 사용하게 된다.

async() 빌더는 코루틴 데이터에 접근하는 수신 함수 await()을 호출할 때 예외를 발생시키고, 디펄트 핸들러가 호출된다. 하위 코루틴을 try - catch블록으로 예외를 처리해도 여전히 프로그램은 예외와 함께 중단된다.

import kotlinx.coroutines.*

fun main() {
    runBlocking {
        val deferredA = async {
            throw Exception("Error in task A")
            println("Task A completed")
        }

        val deferredB = async {
            println("Task B completed")
        }

        try {
            deferredA.await()  // 수신 함수 호출시 예외 발생
            deferredB.await()
        } catch (e: Exception) {
            println("Caught $e")
        }
        println("Root")
    }
}
// result
Caught java.lang.Exception: Error in task A
Root
Exception in thread "main" java.lang.Exception: Error in task A

슈퍼바이저 잡은 취소가 아래 방향으로만 전파된다. 슈퍼바이저 잡이 취소 되면 자동으로 자신의 모든 자식을 취소한다. 하지만 슈퍼바이저가 아니라 자식이 취소되면 슈퍼바이저나 슈퍼바이저의 다른 자식들은 영향을 받지 않는다.

import kotlinx.coroutines.*

fun main() {
    runBlocking {
        supervisorScope {
            val deferredA = async {
                throw Exception("Error in task A")
                println("Task A completed")
            }

            val deferredB = async {
                println("Task B completed")
            }

            try {
               deferredA.await()
               deferredB.await()
            } catch (e: Exception) {
                println("Caught $e")
            }
           
            println("Root")
        }
    }
}
// result
Task B completed
Caught java.lang.Exception: Error in task A
Root

Task B와 루트 코루틴은 영향 없이 출력된다.

13.3 동시성 통신

스레드 안정성을 유지하면서 여러 동시성 작업 사이에 효율적으로 데이터를 공유할 수 있게 해주는 코루틴 라이브러리의 고급 기능 소개 하자.

13.3.1 채널

데이터 스트림을 코루틴 사이의 공유할 수 있는 메커니즘으로, 데이터를 전송하는 send() 메서드와 데이터를 받는 receive() 메서드를 제공한다.

채널 메커니즘

• 자바 Blocking queue와 유사하지만, 스레드를 블록 시키지 않음.
• 양방향 커뮤니케이션에 의한 데이터 스트림을 공유.
  • 채널 버퍼가 꽉 차게 되면 send() 호출이 일시 중단되고, 채널 버퍼가 비어 있을 경우 채널에서 송신할 때까지 receive() 호출은 일시 중단.
• 채널에서 송신한 순서 그대로 원소를 수신되도록 보장. (FIFO)

import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.*
import kotlin.random.Random

fun main() {
    runBlocking {
        val streamSize = 5
        val channel = Channel<Int>(3) // 채널 용량 = 3

        launch {
            for (n in 1..streamSize) {
                delay(Random.nextLong(100))
                val square = n*n
                println("Sending: $square")
                channel.send(square)  // 정수 제곱 값의 스트림을 전송함
            }
        }

        launch {
            for (i in 1..streamSize) {
                delay(Random.nextLong(100))
                val n = channel.receive()
                println("Receiving: $n") // 생성된 스트림을 수신함
            }
        }
    }
}
// result
Sending: 1
Receiving: 1
Sending: 4
Receiving: 4
Sending: 9
Receiving: 9
Sending: 16
Receiving: 16
Sending: 25
Receiving: 25

Channel Buffer Type

• Channel.UNLIMITED (= Int.MAX_VALUE): 채널의 용량은 제한이 없고, 내부 버퍼는 필요에 따라 증가함.
  • 송신 측은 일시 중단 하지 않고, 수신 측의 경우 버퍼가 비어있으면 일시 중단.
• Channel.RENDEZVOUS (= 0): 채널의 버퍼를 설정하지 않을 경우 설정된다. 내부 버퍼가 없는 상태.(default)
  • 송신 측과 수신 측 모두 데이터 스트림이 발생할 때까지 일시 중단 됨.
• Channel.CONFLATED (= -1): 버퍼는 사이즈는 1이고, 송신된 값이 하나로 합쳐지는 채널.
  • 수신 측이 송신 측 보다 처리 속도가 느리거나, 송신 측에 새로운 데이터 스트림이 들어오면 기존의 값을 덮어쓰고 마지막 값을 전송함. 즉, 송신 측은 일시 중단 하지 않음.

Channel <Int>(Channel.RENDEZVOUS)는

val channel = Channel<Int>(Channel.RENDEZVOUS) // 0

//result
Sending: 1
Receiving: 1
Sending: 4
Receiving: 4
Sending: 9
Receiving: 9
Sending: 16
Receiving: 16
Sending: 25
Receiving: 25

값이 합쳐지는 Channel <Int>(Channel.CONFLATED)는 데이터 스트림 중 송신 측에서 일부가 누락되거나 버려 저도 무관할 경우에 적합하다.

import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.*

fun main() {
    runBlocking {
        val streamSize = 5
        val channel = Channel<Int>(Channel.CONFLATED)

        launch {
            for (n in 1..streamSize) {
                delay(100)
                val square = n*n
                println("Sending: $square")
                channel.send(square)
            }
        }

        launch {
            for (i in 1..streamSize) {
                delay(200)             // 소비자 지연 시간 2배
                val n = channel.receive()
                println("Receiving: $n")
            }
        }
    }
}

// result
Sending: 1
Receiving: 1
Sending: 4
Sending: 9
Receiving: 9
Sending: 16
Sending: 25
Receiving: 25

수신자 코루틴이 1부터 streamSize까지 이터레이션 하므로 수신할 것으로 기대하기에 프로그램이 종료되지 않는다. 이 경우 Channle API는 Close() 메서드를 사용해 더 이상 데이터 스트림을 보내지 않는다는 신호를 보낼 수 있다.

소비자 쪽에서는 이터레이션 횟수를 고정하는 대신 채널에서 들어오는 데이터에 대해 이터레이션을 할 수 있다.

launch {
  for (n in 1..streamSize) {
    delay(100)
    val square = n*n
    println("Sending: $square")
    channel.send(square)
  }
  channl.close() // 채널 종료 메서드 호출
}

launch {
	for(n in channel) {
		println("Receiving: $n")
		delay(200)
	}
}

데이터 교환이 완료되면 정상적으로 프로그램이 종료된다.

소비자 쪽에서는 명시적인 이터레이션 대신 consumeEach() 함수를 통해 모든 채널 콘텐츠를 얻어서 사용할 수 있다.

Channle.consumerEach {
	println("Receiving: $it")
	delay(200)
}

consumerEach() 함수는 송신 측 및 수신 측 예외시 CloseSendChannelException 예외가 발생하고, channel 이 닫히고 코루틴이 종료된다.

한 채널의 데이터 스트림을 여러 코루틴이 동시에 receive() 하는 Fan out이라고 한다.

import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.*
import kotlin.random.Random

fun main() {
    runBlocking {
        val streamSize = 5
        val channel = Channel<Int>(2)

        launch {
            for (n in 1..streamSize) {
                val square = n*n
                println("Sending: $square")
                channel.send(square)
            }
            channel.close()
        }

        for (i in 1..3) {  // 3개의 데이터 스트림 소비자 생성
            launch {
                for (n in channel) {
                    println("Receiving by consumer #$i: $n")
                    delay(Random.nextLong(100))
                }
            }
        }
    }
}

// result
Sending: 1
Sending: 4
Sending: 9
Receiving by consumer #1: 1
Receiving by consumer #2: 4
Receiving by consumer #3: 9
Sending: 16
Sending: 25
Receiving by consumer #2: 16
Receiving by consumer #1: 25

위와 반대의 케이스로 여러 생산자 코루틴이 한 채널에 데이트 스트림을 전송하고 한 명의 수신 측 코루틴이 소비하는 Fan in도 있다. 일반적인 경우 여러 생산자와 여러 소비자가 여러 채널을 공유할 수도 있다.

13.3.2 생산자

produce()라는 코루틴 빌더를 사용해 동시성 데이터 스트림을 생성할 수 도 있다. Channel과 비슷한 send() 메서드를 제공하는 ProducerScope 영역을 도입해 준다.

import kotlinx.coroutines.channels.*
import kotlinx.coroutines.*

fun main() {
    runBlocking {
        val channel = produce {
            for (n in 1..5) {
                val square = n*n
                println("Sending: $square")
                send(square)
            }
        }

        launch {
            channel.consumeEach { println("Receiving: $it") }
        }
    }
}
// result
Sending: 1
Receiving: 1
Sending: 4
Sending: 9
Receiving: 4
Receiving: 9
Sending: 16
Sending: 25
Receiving: 16
Receiving: 25

이 경우 채널을 명시적으로 닫을 필요가 없고, 코루틴이 종료되면 produce() 빌더가 채널을 자동으로 닫아준다. 예외 처리 관점에서 produce()는 async()/await() 정책을 따른다.

13.3.3 티커(obsolete)

Ticker Channel은 특별한 채널로, 한 원소와 다음 원소의 발생 시점 이후에 delay 시간이 지나면 Unit 값을 송신한다.

• delayMillis: 티커 원소의 발생 시간 간격을 밀리초 단위로 지정한다.
• initalDelayMillis: 티커 생성 시점과 원소가 최초로 발생하는 시점 사이의 시간 간격이다.
• Context: 티커를 실행할 코루틴 문맥이다
• Mode: 티커의 행동을 결정하는 TickerMode Enum이다.
  • TickerMode.FIXED_PERIOD: 생성되는 원소 사이의 시간 간격을 지정된 지연 시간에 최대한 맞추기 위해 실제 지연 시간을 조정한다.
  • TickerMode.FIXED_DELAY: 실제 흘러간 시간과 관계없이 delayMillis로 지정한 지연 시간만큼 시간을 지연시킨 후 다음 원소를 송신한다.

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

fun main() = runBlocking {
    val ticker = ticker(100)  // 100ms 마다 send()
    println(withTimeoutOrNull(50) { ticker.receive() }) // ticker delayMillis 100 이기에 delay 시간은 50ms 이라 titmeout 발생..
    println(withTimeoutOrNull(60) { ticker.receive() }) // 100ms 지남..
    delay(250)  // 250ms 일시 중단 하는데, 100ms 지나면 티커는 send를 1번 출한다. 그리고, 티커와 소비자 모두 150ms 동안 일시 중단 상태..
    println(withTimeoutOrNull(1) { ticker.receive() })  // 신호가 이미 보내졌기에 결과 반환...
    println(withTimeoutOrNull(60) { ticker.receive() }) // 그리고..250ms 가 지났다는 사실을 티커는 알게되서 딜레이시 시간을 100-50 = 50으로 조정 하기에 60ms 이내로 다시 출력할수 있게 됨.
    println(withTimeoutOrNull(60) { ticker.receive() }) // 이후 티커는 100ms을 대기하기에 60 이내에 응답을 못주고, 결국 못기다리고 timeout
}
// result
null
kotlin.Unit
kotlin.Unit
kotlin.Unit
null

현재 실험 단계에 있으며 미래 코루틴 라이브러리 버전에서는 언제든 변경될 수 있다.

13.3.4 액터(obsolete)

가변 상태를 스레드 안전하게 공유하는 방법을 구현하는 일반적인 방법으로, 코루틴에서 스레드 간에 메시지를 보내서 동시성 통신을 진행할 수 있는 수단을 제공하는 객체이다.

액터는 자기 자신에게 들어오는 메시지를 listen 하고, 자신의 상태를 바꾸면서 메시지에 응답할 수 있으며, 다른 메시지를 보낼 수 있고 새로운 액터를 시작할 수 있다. 결과를 생성하지 않고 Job을 시작한다는 점은 launch() 빌더와 비슷하다.

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.SendChannel
import kotlinx.coroutines.channels.actor

sealed class AccountMessage

class GetBalance(
    val amount: CompletableDeferred<Long>
) : AccountMessage()

class Deposit(val amount: Long) : AccountMessage()

class Withdraw(
    val amount: Long,
    val isPermitted: CompletableDeferred<Boolean>
) : AccountMessage()

fun CoroutineScope.accountManager(
    initialBalance: Long
) = actor<AccountMessage> {

    var balance = initialBalance

    for (message in channel) {
        when (message) {
            is GetBalance ->message.amount.complete(balance) // 액터 클라이언트에 결과 전달

            is Deposit -> {
                balance += message.amount
                println("Deposited ${message.amount}")
            }

            is Withdraw -> {
                val canWithdraw = balance >= message.amount
                if (canWithdraw) {
                    balance -= message.amount
                    println("Withdrawn ${message.amount}")
                }
                message.isPermitted.complete(canWithdraw)
            }
        }
    }
}

액터는 데이터를 받기 위해 채널을 사용하고, 기본적으로 rendezvous 채널을 사용한다. 액터 클라이언트에게 요청 결과를 돌려줄 때는 CompletableDeferred에서 complete() 메서드를 사용한다.

13.4 자바 동시성 사용하기

코틀린에서만 사용할 수 있는 코루틴 라이브러리 외에도 JVM 플랫폼에서는 JDK가 제공하는 동기화 요소를 활용할 수 있다.

13.4.1 스레드 시작하기

범용 스레드를 시작하려면, 스레드에서 실행하려는 실행 가능 객체에 대응하는 람다와 스레드 프로터피들을 지정해서 thread() 함수를 사용하면 된다.

• start: 스레드를 생성하자마자 시작할지 여부(디펄트 true)
• isDaemon: 스레드를 데몬 모드로 시작할지 여부(디펄트 false)
• contextClassLoader: 스레드 코드가 클래스와 자원을 적재할 때 사용할 클래스 로더(디펄트 null)
• name: 커스텀 스레드 이름(디폴트 null), Thread-1..... 등
• priority: Thread.MIN_PRIORITY(=1) ~(=10) 사이의 값으로 정해지는 우선순위로 어떤 스레드가 얼마나 많은 CPU 시간을 배정받는지 결정.
• block: () -> Unit 타입의 함숫값으로 새 스레드가 생성되면 실행할 코드

// 150ms 마다 메시지 출력하는 스레드
import kotlin.concurrent.thread

fun main() {
    println("Starting a thread...")

    thread(name = "Worker", isDaemon = true) { // setDaemon(true)
        for (i in 1..5) {
            println("${Thread.currentThread().name}: $i")
            Thread.sleep(150)
        }
    }

    Thread.sleep(500)
    println("Shutting down...")
}
// result
Worker: 5 Starting a thread...
Worker: 1
Worker: 2
Worker: 3
Worker: 4
Shutting down...

// timer() 함수로도 고정된 시간 간격으로 thread 실행 가능함
timer(period = 150, name = "Worker", daemon = true) {
	println("${Thread.currentThread().name}: ${++counter}")
}

13.4.2 동기화와 락

동기화는 특정 코드 블록이 한 스레드에서만 실행되는 것을 보장한다. 이런 코드 블록을 다른 스레드가 실행하고 있다면 이를 사용하려고 시도하는 스레드는 모두 대기 한다.

자바와 동일하게 코틀린도 동기화 블럭을 사용해 동기화해야 하는 구문 코드를 감싸주면 된다.

import kotlin.concurrent.thread

fun main() {
    var counter = 0
    val lock = Any()

    for (i in 1..5) {
        thread(isDaemon = false) {
            synchronized(lock) {
                counter += i // thread 가 연산 중일 때 다른 스레드는 대기..
                println(counter)
            }
        }
    }
}
// result
6 
10 
13 
15 
1

synchronized block이 없으면
??

일반적으로 synchronized() 함수는 람다의 반환값을 반환한다. 호출 시점의 중간 카운터를 읽을 수도 있다.

val currentCount = synchronized(lock) {counter}
println("count $currentCount")

자바에서는 메서드에 synchronized 변경자를 지정하여 동기화를 구현할 수도 있다. 코틀린에서는 @Synchronized 애너테이션으로 지정할 수 있다.

import kotlin.concurrent.thread

class Counter {
    private var value = 0
    @Synchronized fun addAndPrint(value: Int) {
        this.value += value
        println(value)
    }
}

fun main() {
    val counter = Counter()
    for (i in 1..5) {
        thread(isDaemon = false) { counter.addAndPrint(i) }
    }
}

// result
4 2 5 3 1

코틀린 코루틴 기반 동시성의 기초를 다뤘다. 일시 중단 함수를 통해 동시성 코드를 작성하는 방법 및 코루틴 빌더와 영역 및 문맥을 사용해 코루틴의 생명 주기를 관리하는 방법을 이해했다.

13장 동시성

배우는 내용

• 코루틴
• 동시성 통신
• 자바 동시성 사용

13.1 코루틴

kotlin에서 동시성 처리를 위한 메커니즘으로 명령형 스타일의 코드로 작성해도 컴파일러가 코드를 효율적인 비동기 계산으로 자동변환한다. 즉, 일시 중단 가능한 함수 개념을 가지고 있다.

자바의 동시성 처리 단점?

• 멀티 스레드로 동시성 처리가 가능하지만 블러킹 연산(sleep, join, wait)
• 스레드를 블록하고 실행 재개시 문맥 전환 비용, 스레드마다 시스템 자원이 많이 소모
• 비동기 연산을 위한 다양한 라이브러리가 있지만 일반적인 명령형 흐름제어가 아니라 복잡도가 높아짐(callback)

13.1.1 코루틴과 일시 중단 함수

코루틴의 라이브러리를 뒷받침하는 기본 요소는 일시 중단 함수

• 코루틴의 핵심은 일시 중단 함수 개념
• suspend 변경자로 표기
• 여러 가지 코루틴 빌더(launch, async, runBlocking)

suspend fun kotlin() {
  println("Hello")
  delay(100L) // 일시중단 함수 (스레드를 블럭시키지 않고 다른 작업 수행)
  println("World")
}

일반 함수에서는 일시 중단 함수를 호출할 수 없다.

fun kotlin() {
  println("Hello")
  delay(100L) // error: delay is a suspend function
  println("World")
}

13.1.2 코루틴 빌더

코루틴 빌더들

• launch { } : Job을 return 하는 coroutine
• async { } : Deffered 를 이용하여 single value 를 return 하는 coroutine. Deferred value 에 .await() 를 호출하여 최종 결과를 얻을 수 있다. Deferred 는 Job 이라, cancel 할 수도 있다.
• runBlocking { } : current thread 를 block 시키는 coroutine

launch vs async 비교

함수결과 반환반환 타입

launch	X	Job
async	O	Deferd<T>

launch() 빌더는 코루틴을 시작하고, 코루틴을 실행 중인 작업의 상태를 추적하고 변경할 수 있는 Job 객체를 반환한다.

• 아래 프로그램의 결과를 보면 병렬적으로 실행된 것을 확인 가능
• 다만, 실행 순서가 항상 일정하지는 않음

import kotlinx.coroutines.GlobalScope
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import java.lang.System.currentTimeMillis
​
fun main() { // main이 suspend 함수가 아님에 유의
    val time = currentTimeMillis()
​
    GlobalScope.launch {
        delay(100) // thread 블럭 시키지 않고 코드 수행만 멈추는 특별한 suspending function.
        println("Task 1 finished in ${currentTimeMillis() - time} ms")
    }
​
    GlobalScope.launch {
        delay(100)  
        println("Task 2 finished in ${currentTimeMillis() - time} ms")
    }
    // 스레드를 블럭
    Thread.sleep(200) 
}
​
// result
Task 2 finished in 176 ms
Task 1 finished in 176 ms

async() 빌더는 결과를 Deferred 인스턴스로 돌려주고, await() 메서드를 통해 계산 결과에 접근할 수 있다.

suspend fun main() {
    val message = GlobalScope.async {
        delay(100)
        "abc"
    }
​
    val count = GlobalScope.async {
        delay(100)
        1 + 2
    }
​
    delay(200)
    // await() 실행되면 코루틴은 일시중단되고 결과가 반환되면 코루틴은 다시 재개되고 결과를 출력
    val result = message.await().repeat(count.await())   // await()는 일시중단 가능한 코루틴 내부에서만 사용 가능
    println(result)
}
​
// result 
abcabcabc

Launch() 와 async() 빌더는 일시 중단 함수 내부에서 스레드 호출을 블럭시키지는 않지만, 백그라운드 스레드를 공유하는 풀을 통해 작업을 실행한다.

runBlocking() 빌더는 디폴트로 현재 스레드에서 코루틴을 만들고 코루틴이 완료될 때까지 현재 스레드의 실행을 블럭시킨다.

• 코루틴이 끝나면 일시 중단 람다의 결과값이 된다.
• 취소되면 예외를 발생.
• 블러킹 호출과 넌블러킹 호출 사이의 다리 역할을 위해 사용
• 테스트나 메인 함수에서 최상위 빌더로만 사용해야 함

import kotlinx.coroutines.*
​
fun main() {
    // 공유풀 백그라운드 스레드를 새로 할당 받아 실행
    GlobalScope.launch {
        delay(100)
        println("Background task: ${Thread.currentThread().name}")
    }
    // 메인 스레드에서 실행되고 메인 스레드를 블럭 시킴
    runBlocking {
        println("Primary task: ${Thread.currentThread().name}")
        delay(200)
    }
}
​
// result
Primary task: main @coroutine#2
Background task: DefaultDispatcher-worker-1 @coroutine#1

13.1.3 코루틴 영역과 구조적 동시성

동시성 작업 사이의 부모 자식 관계로 인해 이런 실행 시간 제한이 가능하다. 어떤 코루틴을 다른 코루틴의 문맥에서 실행하면 후자가 전자의 부모가 된다. 이경우 자식의 실행이 모두 끝나야 부모가 끝날 수 있도록 부모와 자식의 생명 주기가 연관된다. 이런 기능을 구조적 동시성(structured concurrency)이라고 한다.

import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
​
fun main() {
    runBlocking {  // 부모 코루틴
        println("Parent task started ${Thread.currentThread().name}")
​
        launch { // 자식 코루틴
            println("Task A started ${Thread.currentThread().name}")
            delay(200)
            println("Task A finished ${Thread.currentThread().name}")
        }
​
        launch { // 자식 코루틴
            println("Task B started ${Thread.currentThread().name}")
            delay(200)
            println("Task B finished ${Thread.currentThread().name}")
        }
​
        delay(100)
        println("Parent task finished ${Thread.currentThread().name}")
    }
    println("Shutting down... ${Thread.currentThread().name}")
}
​
​
// result
Parent task started main @coroutine#1
Task A started main @coroutine#2
Task B started main @coroutine#3
Parent task finished main @coroutine#1 // 지연이 100이기에 runBlocking() 호출의 일시 중단 람다 본문이 먼저 출력됨
Task A finished main @coroutine#2
Task B finished main @coroutine#3
Shutting down... main

최상위 코루틴이 시작하고 CoroutineScope 인스턴스 안에서 launch를 호출해 두가지 자식 코루틴을 시작한다.

실행 순서 : 부모 runBlocking 코루틴 -> 각 자식 launch 코루틴 시작 -> 자식 200 delay -> 부모 runBlocking 코루틴 실행 -> 각 자식 launch 의 코루틴 출력 -> 자식 코루틴이 종료될 때까지 기다린 후 runBlocking 실행 -> 메인 스레드 블럭 해제 후 main 출력되고 종료

coroutineScope()와 runBlocking()의 가장 큰 차이는 coroutineScope()가 일시 중단 함수라 현재 스레드를 블럭시키지 않는다는 점이다.

import kotlinx.coroutines.*
​
fun main() {
    runBlocking {  // 부모 코루틴
        println("Custom scope start")
​
        coroutineScope {  // 아래 자식 코루틴 실행이 끝날 때까지 일시 중단 처리 됨
            launch {  // 자식 코루틴
                delay(100)
                println("Task 1 finished")
            }
​
            launch {  // 자식 코루틴
                delay(100)
                println("Task 2 finished")
            }
        }
​
        println("Custom scope end")
    }
}
​
// result
Custom scope start main @coroutine#1
Task 1 finished main @coroutine#2
Task 2 finished main @coroutine#3
Custom scope end main @coroutine#1

13.1.4 코루틴 문맥

코루틴은 항상 특정 문맥에서 실행되는데 코루틴을 감싸는 변수 영역의 CoroutineContext 프로퍼티를 통해 이 문맥에 접근할 수 있고, 키-값 쌍으로 이뤄진 불변 컬렉션이다.

GlobalScope.launch {
  // 현재 잡을 얻고 "Task is active: true" 를 출력
  println("Task is active: $coroutineContext[Job.Key]!!.isActive")
}

13.2 코루틴 흐름 제어와 잡 생명 주기

잡은 동시성 작업의 생명 주기를 표현하는 객체다. 잡을 사용하면 작업 상태를 추적하고 필요할 때 작업을 취소할 수 있다.

Active: 작업이 시작됐고 아직 완료나 취소로 끝나지 않은 상태로, 보통 디폴트 상태.

• CoroutineStart.DEFAULT는 디폴트 동작이며, 잡을 즉시 시작한다.
• CoroutineStart.LAZY는 잡을 자동으로 시작하지 말라는 뜻으로, 잡이 New 상태가 되고 시작을 기다리게 된다.

Completing: 코루틴이 일시 중단 람다 블록의 실행을 끝낸 잡의 상태(완료 중)

• 자식들의 완료를 기다리는 상태로 잡은 모든 자식이 완료될 때까지 이 상태를 유지

Completed: 모든 자식이 완료된 상태(완료 됨)

New 상태의 잡에 대해 start() or join() 메서드를 호출하면 잡이 시작되면서 활성화 상태가 된다.

import kotlinx.coroutines.*

fun main() {
    runBlocking {  // 부모 코루틴
        val job = launch(start = CoroutineStart.LAZY) {  // 자식 코루틴
            println("Job started")
        }

        delay(100)

        println("Preparing to start...")
        job.start() // start 메서드가 호출되면 자식 코루틴 시작..
    }
}

// result
Preparing to start...
Job started

자식 코루틴의 시작을 부모 코루틴이 메세지를 호출한 뒤로 미룬다.

잡의 join() 메서드를 사용하면 조인 대상 잡이 완료될 때까지 현재 코루틴을 일시 중단시킬 수 있다. 두 자식 메세지의 실행이 끝난 후에 루트 코루틴이 출력되도록 보장해 준다.

import kotlinx.coroutines.*

fun main() {
    runBlocking {
        val job = coroutineContext[Job.Key]!!
        val jobA = launch { println("This is task A") }
        val jobB = launch { println("This is task B") }

        jobA.join()
        jobB.join()
				
        println("${job.children.count()} children running")
    }
}

// result
This is task A
This is task B
0 children running  

현재 잡의 상태를 추적할 수 있는 프로퍼티들.

상태가 "완료됨, "취소됨"인 잡의 isCompleted가 true이기에, Cancelled와 Completed는 isCancelled로 구분 할 수 있다.

13.2.1 취소

잡의 cancel() 메서드를 호출하면 잡을 취소할 수 있다. 취소에는 협력이 필요하다. 즉, 취소 가능한 코루틴이 스스로 취소가 요청됐는지 검사해서 적절히 반응해줘야 한다.

import kotlinx.coroutines.*

// 코루틴을 취소했지만 취소를 위해 협력되지 않기에 취소 불가함
suspend fun main() {
    val squarePrinter = GlobalScope.launch(Dispatchers.Default) {
        var i = 1
        while (true) {
            println(i++)
        }
    }

    delay(100) // 자식 잡이 어느 정도 실행될 시간을 준다
    squarePrinter.cancel()
}

// isActive 프로퍼티를 현재 잡의 활성화 상태
suspend fun main() {
    val squarePrinter = GlobalScope.launch(Dispatchers.Default) {
        var i = 1
        while (isActive) {  // 취소 가능한 형태의 루프
            println(i++)
        }
    }

    delay(100) 
    squarePrinter.cancel()  // 부모 코루틴이 취소되면 자동으로 모든 자식의 실행을 취소
}

isActive 프로퍼티는 CoroutineScope 객체가 자체적으로 지원하는 프로퍼티로, 코루틴의 취소 상태를 Boolean 타입으로 갖고 있다. 부모 코루틴이 cancel() 메서드를 호출하면 squarePrinter의 상태가 취소 중으로 바뀌고, 그 다음 isActive 검사를 통해 루프를 종료 시킬 수 있다.

import kotlinx.coroutines.*

fun main() = runBlocking {
    val job = launch {
        repeat(1000) { i ->
            println("${Thread.currentThread().name} : 코루틴아 달려라 $i ...")
            delay(300L)
        }
    }

    delay(1000L)
    println("${Thread.currentThread().name} : 코루틴아 이제 그만 쉬어")

    //job.cancel()  // Job 취소
    //job.join()    // Job 끝날 때 까지 대기
    job.cancelAndJoin()  // 취소하고 끝날 때 까지 대기
    println("${Thread.currentThread().name} : 코루틴 종료")
}

// result
main @coroutine#2 : 코루틴아 달려라 0 ...
main @coroutine#2 : 코루틴아 달려라 1 ...
main @coroutine#2 : 코루틴아 달려라 2 ...
main @coroutine#2 : 코루틴아 달려라 3 ...
main @coroutine#1 : 코루틴아 이제 그만 쉬어
main @coroutine#1 : 코루틴 종료

13.2.2 타임아웃

작업이 완료되기를 무작정 기다리 수 없어서 타임아웃을 설정이 필요할 경우

• withTimeout() : TimeoutCancellationException 반환
• withTimeoutOrNull() : 예외 대신 null 반환

import kotlinx.coroutines.*
import java.io.File

fun main() {
    runBlocking {
        val asyncData = async { File("data.txt").readText() }
        try {
            val text = withTimeout(50) { asyncData.await() }
            println("Data loaded: $text")
        } catch (e: Exception) {
            println("Timeout exceeded")
        }
    }
}

13.2.3 코루틴 디스패치하기

코루틴은 스레드와 무관하게 일시 중단 가능한 계산을 구현할 수 있게 해주지만, 코루틴을 실행하려면 여전히 스레드와 연관시켜야 한다. 코루틴 라이브러리에는 특정 코루틴을 실행할 때 사용할 스레드를 제어하는 특별한 코루틴 디스패처 컴포넌트가 있다.

• 디스패처는 코루틴 문맥의 일부
• launch() or runBlocking() 등에서 지정할 수 있음

import kotlinx.coroutines.*

fun main() {
  runBlocking {
    // 전역 스레드 풀 디스패처를 사용해 코루틴을 실행한다
    launch(Dispatchers.Default) {
      println(Thread.currentThread().name) // DefaultDispatcher-worker-1
    }
    println(Thread.currentThread().name) // main
  }
}

코루틴 디스패처는 병렬 작업 사이에 스레드를 배분해 주는 자바 실행기(executor)와 비슷함.

스레드 개수를 지정하려면 asCoroutineDispatcher() 확장 함수를 사용하여 코루틴 문맥으로 전환 할 수 있다.

import kotlinx.coroutines.*
import java.util.concurrent.ScheduledThreadPoolExecutor
import java.util.concurrent.atomic.AtomicInteger

fun main() {
    val id = AtomicInteger(0)

    val executor = ScheduledThreadPoolExecutor(5) { runnable ->
        Thread(
            runnable,
            "WorkerThread-${id.incrementAndGet()}"
        ).also { it.isDaemon = true }  // 코루틴 종료 후 자동으로 종료
    }

    executor.asCoroutineDispatcher().use { dispatcher ->
        runBlocking {
            for (i in 1..6) {
                launch(dispatcher) {
                    println(Thread.currentThread().name)
                    delay(1000)
                }
            }
        }
    }
}

// result
WorkerThread-1 @coroutine#2
WorkerThread-4 @coroutine#5
WorkerThread-2 @coroutine#3
WorkerThread-3 @coroutine#4
WorkerThread-5 @coroutine#6
WorkerThread-2 @coroutine#7

asCoroutineDispatcher()를 호출하면 ExecutorCoroutineDispatcher를 반환하고, Closeable 인스턴스도 구현하기에 시스템 자원을 해제하려면 close() or use() 함수로 해제가 가능하다.

코루틴 라이브러리에서 제공하는 디스패처

• Dispatchers.Default: shared common pool 기반으로 pool 크기는 디폴트로 사용 가능한 CPU 코어 수이거나 일반적인 background job을 수행
• Dispatchers.IO: shared pool 기반으로 파일을 읽고 쓰는 것처럼 잠재적으로 블러킹될 수 있는 I/O 작업에 최적화.
• Dispatchers.Main: 사용자 입력이 처리되는 UI 스레드에서 사용.(android view..)

newFixedThreadPoolContext() or newSingleThreadPoolContext()를 사용하여 새로운 스레드를 생성하거나 사용하는 디스패처를 만들 수도 있다. 다만, 실험적인 기능이라 추후 대치될 예정.

디스패처를 명시적으로 지정하지 않으면 코루틴이 시작한 영역부터 자동으로 상속된다.

import kotlinx.coroutines.*

fun main() {
    runBlocking { // main
        println("Root: ${Thread.currentThread().name}")

        launch {  // main
            println("Nested, inherited: ${Thread.currentThread().name}")
        }
				// 코루틴 문맥이 실행될 스레드 지정
        launch(Dispatchers.Default) {  // DefaultDispatcher-worker-1
            println("Nested, explicit: ${Thread.currentThread().name}")
        }
    }
}

// result
Root: main @coroutine#1
Nested, inherited: main @coroutine#2
Nested, explicit: DefaultDispatcher-worker-1 @coroutine#3

부모 코루틴이 없으면 암시적으로 Dispatchers.Default로 디스패처를 가정한다. 다만 runBlocking() 빌더는 현재 스레드를 사용한다. 특정 스레드에서만 실행하고 싶을 경우 withContext() 함수를 사용하면 가능하다.

import kotlinx.coroutines.*

@Suppress("EXPERIMENTAL_API_USAGE")
fun main() {
    newSingleThreadContext("Worker").use { worker ->
        runBlocking {
            println(Thread.currentThread().name)   // main
            withContext(worker) {
                println(Thread.currentThread().name) // Worker
            }
            println(Thread.currentThread().name)   // main
        }
    }
}

중단 가능 루틴의 일부를 한 스레드에서만 실행하고 싶을 때 유용하다.

13.2.4 예외 처리

lunch() 빌더는 예외를 부모 코루틴에게 전달하고, CoroutineExceptionHandler에 의해 처리 된다.

• 부모 코루틴이 (자식에게서 발생한 오류와) 똑같은 오류로 취소되고, 나머지 자식도 모두 취소 됨
• 자식들이 모두 취소되고 나면 부모는 예외를 코루틴 트리의 윗부분으로 전달(루트 코루틴)

import kotlinx.coroutines.*

fun main() {
    // 커스텀 핸들러..
    runBlocking {  // 최상위 코루틴
        launch { // 첫번째 자식 코루틴
            throw Exception("Error in task A")
            println("Task A completed")
        }

        launch { // 두번째 자식 코루틴
            delay(1000)
            println("Task B completed")
        }

        println("Root")
    }
}
// result
Root
Exception in thread "main" java.lang.Exception: Error in task A

최상위 코루틴 start -> 첫 번째 코루틴 예외 -> 최상위 코루틴 취소 -> 최상위 하위 자식 코루틴 취소

최상위에서 지정된 핸들러가 없기 때문에 Thread.uncauhtExceptionHandler에 의한 디펄트 동작을 실행 한다.

CoroutineExceptionHandler는 현재 코루틴 문맥과 던져진 예외를 인자로 전달 받는다.

fun handleException(context: CoroutineContext, exception: throwable)

val handler = CoroutineExceptionHandler{ _, exception -> // 인자 2개인 람다를 받는 함수 생성
	println("Caught $exception")
}

코루틴 예외 핸들러의 인스턴스가 예외를 처리하도록 지정하려면 코루틴 문맥에 인스턴스를 넣어야 한다.

import kotlinx.coroutines.*

suspend fun main() {
    val handler = CoroutineExceptionHandler{ _, exception ->
        println("Caught $exception")
    }
    // 전역 영역에서 실행 및 handler 인스턴스 선언
    GlobalScope.launch(handler) {
        launch {
            throw Exception("Error in task A")
            println("Task A completed")
        }

        launch {
            delay(1000)
            println("Task B completed")
        }

        println("Root")
    }.join()
}
// result
Root
Caught java.lang.Exception: Error in task A

CoroutineExceptionHandler는 전역 영역에서 실행된 코루틴에 대해서만 정의할 수 있고, CoroutineExceptionHandler가 정의된 코루틴의 자식에 대해서만 적용된다. 앞선 runBlocking()의 경우 코루틴이 전역 영역에서 실행되지 않아 디펄트 핸들러를 사용하게 된다.

async() 빌더는 코루틴 데이터에 접근하는 수신 함수 await()을 호출 할 때 예외를 발생 시키고, 디펄트 핸들러가 호출 된다. 하위 코루틴을 try - catch블럭으로 예외를 처리 해도 여전히 프로그램은 예외와 함께 중단 된다.

import kotlinx.coroutines.*

fun main() {
    runBlocking {
        val deferredA = async {
            throw Exception("Error in task A")
            println("Task A completed")
        }

        val deferredB = async {
            println("Task B completed")
        }

        try {
            deferredA.await()  // 수신 함수 호출시 예외 발생
            deferredB.await()
        } catch (e: Exception) {
            println("Caught $e")
        }
        println("Root")
    }
}
// result
Caught java.lang.Exception: Error in task A
Root
Exception in thread "main" java.lang.Exception: Error in task A

슈퍼바이저 잡은 취소가 아래 방향으로만 전파 된다. 슈퍼바이저 잡이 취소 되면 자동으로 자신의 모든 자식을 취소한다. 하지만 슈퍼바이저가 아니라 자식이 취소되면 슈퍼바이저나 슈퍼바이저의 다른 자식들은 영향을 받지 않는다.

import kotlinx.coroutines.*

fun main() {
    runBlocking {
        supervisorScope {
            val deferredA = async {
                throw Exception("Error in task A")
                println("Task A completed")
            }

            val deferredB = async {
                println("Task B completed")
            }

            try {
               deferredA.await()
               deferredB.await()
            } catch (e: Exception) {
                println("Caught $e")
            }
           
            println("Root")
        }
    }
}
// result
Task B completed
Caught java.lang.Exception: Error in task A
Root

Task B와 루트 코루틴은 영향없이 출력된다.

13.3 동시성 통신

스레드 안정성을 유지하면서 여러 동시성 작업 사이에 효율적으로 데이터를 공유할 수 있게 해주는 코루틴 라이브러리의 고급 기능 소개 하자.

13.3.1 채널

데이터 스트림을 코루틴 사이의 공유할 수 있는 메커니즘으로, 데이터를 전송하는 send() 메서드와 데이터를 받는 receive() 메서드를 제공한다.

채널 메커니즘

• 자바 Blocking queue와 유사하지만, 스레드를 블럭시키지 않음.
• 양방향 커뮤니케이션에 의한 데이터 스트림을 공유.
  • 채널 버퍼가 꽉차게 되면 send() 호출이 일시 중단되고, 채널 버퍼가 비어 있을 경우 채널에서 송신할 때까지 receive() 호출은 일시 중단.
• 채널에서 송신한 순서 그대로 원소를 수신되도록 보장. (FIFO)

import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.*
import kotlin.random.Random

fun main() {
    runBlocking {
        val streamSize = 5
        val channel = Channel<Int>(3) // 채널 용량 = 3

        launch {
            for (n in 1..streamSize) {
                delay(Random.nextLong(100))
                val square = n*n
                println("Sending: $square")
                channel.send(square)  // 정수 제곱 값의 스트림을 전송함
            }
        }

        launch {
            for (i in 1..streamSize) {
                delay(Random.nextLong(100))
                val n = channel.receive()
                println("Receiving: $n") // 생성된 스트림을 수신함
            }
        }
    }
}
// result
Sending: 1
Receiving: 1
Sending: 4
Receiving: 4
Sending: 9
Receiving: 9
Sending: 16
Receiving: 16
Sending: 25
Receiving: 25

Channel Buffer Type

• Channel.UNLIMITED (= Int.MAX_VALUE): 채널의 용량은 제한이 없고, 내부 버퍼는 필요에 따라 증가함.
  • 송신측은 일시 중단 하지 않고, 수신측의 경우 버퍼가 비어있으면 일시 중단.
• Channel.RENDEZVOUS (= 0): 채널의 버퍼를 설정하지 않을 경우 설정된다. 내부 버퍼가 없는 상태.(default)
  • 송신측과 수신측 모두 데이터 스트림이 발생할 때까지 일시 중단 됨.
• Channel.CONFLATED (= -1): 버퍼는 사이즈는 1이고, 송신된 값이 하나로 합쳐지는 채널.
  • 수신측이 송신측 보다 처리 속도가 느리거나, 송신측에 새로운 데이터 스트림이 들어오면 기존의 값을 덮어 쓰고 마지막 값을 전송함. 즉, 송신측은 일시 중단 하지 않음.

Channel<Int>(Channel.RENDEZVOUS)는 생산자와 소비자 코루틴이 교대로 활성화되기에 안정적인 동작 순서를 보장한다.

val channel = Channel<Int>(Channel.RENDEZVOUS) // 0

//result
Sending: 1
Receiving: 1
Sending: 4
Receiving: 4
Sending: 9
Receiving: 9
Sending: 16
Receiving: 16
Sending: 25
Receiving: 25

값이 합쳐지는 Channel<Int>(Channel.CONFLATED)는 데이터 스트림 중 송신측에서 일부가 누락되거나 버려저도 무관할 경우에 적합하다.

import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.*

fun main() {
    runBlocking {
        val streamSize = 5
        val channel = Channel<Int>(Channel.CONFLATED)

        launch {
            for (n in 1..streamSize) {
                delay(100)
                val square = n*n
                println("Sending: $square")
                channel.send(square)
            }
        }

        launch {
            for (i in 1..streamSize) {
                delay(200)             // 소비자 지연 시간 2배
                val n = channel.receive()
                println("Receiving: $n")
            }
        }
    }
}

// result
Sending: 1
Receiving: 1
Sending: 4
Sending: 9
Receiving: 9
Sending: 16
Sending: 25
Receiving: 25

수신자 코루틴이 1부터 streamSize까지 이터레이션 하므로 수신할 것으로 기대하기에 프로그램이 종료되지 않는다. 이 경우 Channle API는 Close() 메서드를 사용해 더이상 데이터 스트림을 보내지 않는 다는 신호를 보낼 수 있다.

소비자쪽에서는 이터레이션 횟수를 고정하는 대신 채널에서 들어오는 데이터에 대해 이터레이션을 할 수 있다.

launch {
  for (n in 1..streamSize) {
    delay(100)
    val square = n*n
    println("Sending: $square")
    channel.send(square)
  }
  channl.close() // 채널 종료 메서드 호출
}

launch {
	for(n in channel) {
		println("Receiving: $n")
		delay(200)
	}
}

데이터 교환이 완료되면 정상적으로 프로그램이 종료된다.

소비자 쪽에서는 명시적인 이터레이션 대신 consumeEach() 함수를 통해 모든 채널 콘텐츠를 얻어서 사용할 수 있다.

Channle.consumerEach {
	println("Receiving: $it")
	delay(200)
}

consumerEach() 함수는 송신측 및 수신측 예외시 CloseSendChannelException 예외가 발생하고, channel 이 닫히고 코루틴이 종료된다.

한 채널의 데이터 스트림을 여러 코루틴이 동시에 receive() 하는 Fan out 이라고 한다.

import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.*
import kotlin.random.Random

fun main() {
    runBlocking {
        val streamSize = 5
        val channel = Channel<Int>(2)

        launch {
            for (n in 1..streamSize) {
                val square = n*n
                println("Sending: $square")
                channel.send(square)
            }
            channel.close()
        }

        for (i in 1..3) {  // 3개의 데이터 스트림 소비자 생성
            launch {
                for (n in channel) {
                    println("Receiving by consumer #$i: $n")
                    delay(Random.nextLong(100))
                }
            }
        }
    }
}

// result
Sending: 1
Sending: 4
Sending: 9
Receiving by consumer #1: 1
Receiving by consumer #2: 4
Receiving by consumer #3: 9
Sending: 16
Sending: 25
Receiving by consumer #2: 16
Receiving by consumer #1: 25

위와 반대의 케이스로 여러 생산자 코루틴이 한 채널에 데이트 스트림을 전송하고 한명의 수신측 코루틴이 소비하는 Fan in도 있다. 일반적인 경우 여러 생산자와 여러 소비자가 여러 채널을 공유할 수도 있다.

13.3.2 생산자

produce() 라는 코루틴 빌더를 사용해 동시성 데이터 스트림을 생성할 수 도 있다. Channel과 비슷한 send() 메서드를 제공하는 ProducerScope 영역을 도입해준다.

import kotlinx.coroutines.channels.*
import kotlinx.coroutines.*

fun main() {
    runBlocking {
        val channel = produce {
            for (n in 1..5) {
                val square = n*n
                println("Sending: $square")
                send(square)
            }
        }

        launch {
            channel.consumeEach { println("Receiving: $it") }
        }
    }
}
// result
Sending: 1
Receiving: 1
Sending: 4
Sending: 9
Receiving: 4
Receiving: 9
Sending: 16
Sending: 25
Receiving: 16
Receiving: 25

이 경우 채널을 명시적으로 닫을 필요가 없고, 코루틴이 종료되면 produce() 빌더가 채널을 자동으로 닫아준다. 예외 처리 관점에서 produce()는 async()/await() 정책을 따른다.

13.3.3 티커(obsolete)

Ticker Channel은 특별한 채널로, 한 원소와 다음 원소의 발생 시점 이후에 delay 시간이 지나면 Unit 값을 송신한다.

• delayMillis: 티커 원소의 발생 시간 간격을 밀리초 단위로 지정한다.
• initalDelayMillis: 티커 생성 시점과 원소가 최초로 발생하는 시점 사이의 시간 간격이다.
• Context: 티커를 실행할 코루틴 문맥이다
• Mode: 티커의 행동을 결정하는 TickerMode Enum이다.
  • TickerMode.FIXED_PERIOD: 생성되는 원소 사이의 시간 간격을 지정된 지연 시간에 최대한 맞추기 위해 실제 지연 시간을 조정한다.
  • TickerMode.FIXED_DELAY: 실제 흘러간 시간과 관계없이 delayMillis로 지정한 지연 시간만큼 시간을 지연시킨 후 다음 원소를 송신한다.

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

fun main() = runBlocking {
    val ticker = ticker(100)  // 100ms 마다 send()
    println(withTimeoutOrNull(50) { ticker.receive() }) // ticker delayMillis 100 이기에 delay 시간은 50ms 이라 titmeout 발생..
    println(withTimeoutOrNull(60) { ticker.receive() }) // 100ms 지남..
    delay(250)  // 250ms 일시 중단 하는데, 100ms 지나면 티커는 send를 1번 출한다. 그리고, 티커와 소비자 모두 150ms 동안 일시 중단 상태..
    println(withTimeoutOrNull(1) { ticker.receive() })  // 신호가 이미 보내졌기에 결과 반환...
    println(withTimeoutOrNull(60) { ticker.receive() }) // 그리고..250ms 가 지났다는 사실을 티커는 알게되서 딜레이시 시간을 100-50 = 50으로 조정 하기에 60ms 이내로 다시 출력할수 있게 됨.
    println(withTimeoutOrNull(60) { ticker.receive() }) // 이후 티커는 100ms을 대기하기에 60 이내에 응답을 못주고, 결국 못기다리고 timeout
}
// result
null
kotlin.Unit
kotlin.Unit
kotlin.Unit
null

현재 실험 단계에 있으며 미래 코루틴 라이브러리 버전에서는 언제든 변경될 수 있다.

13.3.4 액터(obsolete)

가변 상태를 스레드 안전하게 공유하는 방법을 구현하는 일반적인 방법으로, 코루틴에서 스레드간에 메시지를 보내서 동시성 통신을 진행할 수 있는 수단을 제공하는 객체이다.

액터는 자기 자신에게 들어오는 메시지를 listen하고, 자신의 상태를 바꾸면서 메시지에 응답할 수 잇으며, 다른 메시지를 보낼 수 있고 새로운 액터를 시작할 수 있다. 결과를 생성하지 않고 Job을 시작한다는 점은 launch() 빌더와 비슷하다.

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.SendChannel
import kotlinx.coroutines.channels.actor

sealed class AccountMessage

class GetBalance(
    val amount: CompletableDeferred<Long>
) : AccountMessage()

class Deposit(val amount: Long) : AccountMessage()

class Withdraw(
    val amount: Long,
    val isPermitted: CompletableDeferred<Boolean>
) : AccountMessage()

fun CoroutineScope.accountManager(
    initialBalance: Long
) = actor<AccountMessage> {

    var balance = initialBalance

    for (message in channel) {
        when (message) {
            is GetBalance ->message.amount.complete(balance) // 액터 클라이언트에 결과 전달

            is Deposit -> {
                balance += message.amount
                println("Deposited ${message.amount}")
            }

            is Withdraw -> {
                val canWithdraw = balance >= message.amount
                if (canWithdraw) {
                    balance -= message.amount
                    println("Withdrawn ${message.amount}")
                }
                message.isPermitted.complete(canWithdraw)
            }
        }
    }
}

액터는 데이터를 받기 위해 채널을 사용하고, 기본적으로 rendezvous 채널을 사용한다. 액터 클라이언트에게 요청 결과를 돌려줄 때는 CompletableDeferred에서 complete() 메서드를 사용한다.

13.4 자바 동시성 사용하기

코틀린에서만 사용할 수 있는 코루틴 라이브러리 외에도 JVM 플랫폼에서는 JDK가 제공하는 동기화 요소를 활용할 수 있다.

13.4.1 스레드 시작하기

범용 스레드를 시작하려면, 스레드에서 실행하려는 실행 가능 객체에 대응하는 람다와 스레드 프로터피들을 지정해서 thread() 함수를 사용하면 된다.

• start: 스레드를 생성하자마자 시작할지 여부(디펄트 true)
• isDaemon: 스레드를 데몬 모드로 시작할지 여부(디펄트 false)
• contextClassLoader: 스레드 코드가 클래스와 자원을 적재할 때 사용할 클래스 로더(디펄트 null)
• name: 커스텀 스레드 이름(디폴트 null), Thread-1.....등
• priority: Thread.MIN_PRIORITY(=1) ~(=10) 사이의 값으로 정해지는 우선순위로 어떤 스레드가 얼마나 많은 CPU 시간을 배정받는지 결정.
• block: () -> Unit 타입의 함숫값으로 새 스레드가 생성되면 실행할 코드

// 150ms 마다 메시지 출력하는 스레드
import kotlin.concurrent.thread

fun main() {
    println("Starting a thread...")

    thread(name = "Worker", isDaemon = true) { // setDaemon(true)
        for (i in 1..5) {
            println("${Thread.currentThread().name}: $i")
            Thread.sleep(150)
        }
    }

    Thread.sleep(500)
    println("Shutting down...")
}
// result
Worker: 5 Starting a thread...
Worker: 1
Worker: 2
Worker: 3
Worker: 4
Shutting down...

// timer() 함수로도 고정된 시간 간격으로 thread 실행 가능함
timer(period = 150, name = "Worker", daemon = true) {
	println("${Thread.currentThread().name}: ${++counter}")
}

13.4.2 동기화와 락

동기화는 특정 코드 블럭이 한 스레드에서만 실행되는 것을 보장한다. 이런 코드 블럭을 다른 스레드가 실행하고 있다면 이를 사용하려고 시도하는 스레드는 모두 대기 한다.

자바와 동일하게 코틀린도 동기화 블럭을 사용해 동기화 해야하는 구문 코드를 감싸주면 된다.

import kotlin.concurrent.thread

fun main() {
    var counter = 0
    val lock = Any()

    for (i in 1..5) {
        thread(isDaemon = false) {
            synchronized(lock) {
                counter += i // thread 가 연산 중일 때 다른 스레드는 대기..
                println(counter)
            }
        }
    }
}
// result
6 
10 
13 
15 
1

synchronized block이 없으면
??

일반적으로 synchronized() 함수는 람다의 반환값을 반환한다. 호출 시점의 중간 카운터를 읽을 수도 있다.

val currentCount = synchronized(lock) {counter}
println("count $currentCount")

자바에서는 메서드에 synchronized 변경자를 지정하여 동기화를 구현할 수도 있다. 코틀린에서는 @Synchronized 애너테이션으로 지정할 수 있다.

import kotlin.concurrent.thread

class Counter {
    private var value = 0
    @Synchronized fun addAndPrint(value: Int) {
        this.value += value
        println(value)
    }
}

fun main() {
    val counter = Counter()
    for (i in 1..5) {
        thread(isDaemon = false) { counter.addAndPrint(i) }
    }
}

// result
4 2 5 3 1

코틀린 코루틴 기반 동시성의 기초를 다뤘다. 일시 중단 함수를 통해 동시성 코드를 작성하는 방법 및 코루틴 빌더와 영역 및 문맥을 사용해 코루틴의 생명 주기를 관리하는 방법에 대하여 정리하였습니다.

참고 자료

https://kotlinlang.org/docs/home.html

 

Kotlin Docs | Kotlin

 

kotlinlang.org

https://play.kotlinlang.org/

 

Kotlin Playground: Edit, Run, Share Kotlin Code Online

 

play.kotlinlang.org

https://myungpyo.medium.com/reading-coroutine-official-guide-thoroughly-part-0-20176d431e9d

 

Reading Coroutine official guide thoroughly — Part 0

요즘 안드로이드 앱 개발자들은 주개발 언어가 Java 에서 Kotlin 으로 많이 전환 되고 있다고 생각합니다. iOS 가 objective-c 에서 swift 로 전환할 때 처럼 안드로이드 SDK 공식 가이드에서도 Kotlin 과 Java

myungpyo.medium.com