[Kotlin] 동시성 프로그래밍

동시성 프로그래밍

• 동기적 수행

  순서대로 작업을 수행하여 하나의 루틴을 완료한 후 다른 루틴을 실행하는 방식

  다양한 기능이 한꺼번에 일어나는 다중 실행 환경에서는 성능상의 제약이 있음

  ex) UI작업, 데이터 다운로드 등을 동시에 대응해야 하는 경우

• 비동기적 수행

  다양한 기능을 동시에 수행할 수 있는 방식

  coroutines, 스레드를 이용하거나 RxJava, Reactive와 같은 서드파티 라이브러리에서 제공

코루틴

: 하나의 개별적인 작업(routine) + 여러 개의 루틴들이 협력(co)한다는 의미로 만들어진 합성어

순차적인 코드처럼 보이지만 비동기 효과를 낼 수 있음 !

코루틴을 왜 사용하는가?

안드로이드에서 서버에서 데이터 통신할때 필요한 많은 과정들.. (UI스레드에서 데이터를 불러오면 x, 무수한 callback 과정들 생략 가능 )

태스크(Task)

: 큰 실행 단위인 프로세스나 좀 더 작은 실행 단위인 스레드로 생각할 수 있음

• 프로세스

  실행되는 메모리, 스택, 파일 등을 모두 포함하기 때문에 프로세스간 문맥 교환(context-switching)을 하는데 비용이 큼

• 스레드

  자신의 스택만 독립적으로 가지고 나머지는 대부분 공유하므로 문맥 교환 비용이 낮아서 자주 사용됨, 하지만 너무 여러개의 스레드를 구성하면 복잡해진다.

  1. Thread 클래스를 이용한 방법

      class SimpleThread : Thread() {
      	override fun run() {
      		println("Class Thread ${Thread.currentThread()}")
      	}
      }
     
      fun main() {
      	val thread = SimpleThread()
      	thread.start()
      }
     
      //익명객체를 사용
      object : Thread() {
      	override fun run() {
      		println("object Thread: ${Thread.curretnThread()}")
      	}
      }.start()
     
      //람다식 사용
      Thread {
      	println("Lambda Thread: ${Thread.currentThread()}")
      }.start() 
     

  2. 인터페이스를 이용한 방법

    class SimpleRunnable : Runnable {
    	override fun run() {
    		println("Interface Thread ${Thread.currentThread()}")
    	}
    }
  
    fun main() {
    	val runnable = SimpleThread()
    	val thread2 = Thread(runnable)
    	thread2.start()
    }
  

문맥 교환(context-switching)

: 하나의 프로세스나 스레드가 CPU를 사용하고 있는 상태에서 다른 프로세스나 스레드가 CPU를 사용하도록 하기 위해, 이전의 프로세스의 상태를 보관하고 새로운 프로세스의 상태를 적재하는 과정

→ 스레드 안에 여러개의 코루틴을 사용한다면, 문맥 교환이 일어나지 않음

---

코루틴(Coroutines)

: 복잡성을 줄이고도 손쉽게 일시 중단하거나 다시 시작하는 루틴을 만들 수 있음

문맥 교환 없이, 해당 루틴을 일시 중단(suspended)를 통해 제어

스레드와 다르게 스택을 가지지 않으므로 생성 오버헤드가 줄어듦

• 코루틴 빌더

  • launch

    -일단 실행하고 잊어버리는 형태의 코투린, 메인 프로그램과 독립되어 실행할 수 있다.

    -즉시 실행하며 블록 내의 실행 결과는 반환 x

    -상위 코드를 블록 시키지 않고, 관리를 위한 **Job객체를 즉시 반환 **

    -join을 통해 상위 코드가 종료되지 않고 완료를 기다리게 할 수 있다.

  • async

    -결과나 예외를 반환함

    -**Deffered를 통해 실행 결과를 반환하며, await을 통해 받을 수 있다**.

    -await은 작업이 완료될 때 까지 기다리게 됨

  • runBlocking

    자신을 호출한 스레드를 blocking 함

import kotlinx.coroutines.*

fun main() {
    GlobalScope.launch { //새로운 코루틴 백그라운드에서 실행 
        delay(1000L)//1초동안 넌블로킹 지연 -> main은 그대로 쭉 실행 
        pritnln("World!")//2초 지연 되는 동안 출력 2  
    }
    println("Hello ")//먼저 출력한 후 1
    Thread.sleep(2000L)//아직 끝나지 않은 코루틴을 기다리기 위해 2초 지연 
}

• delay()

    public suspend fun delay(timeMillis: kotlin.Long): kotlin.Unit { ... }
  

  : 컴파일러는 suspend가 붙은 함수를 자동적으로 추출해 분리된 루틴을 만듦

• Job

  코루틴의 생명주기를 관리하며 생성된 코루틴 작업들은 부모-자식과 같은 관계를 가질 수 있음

  -부모가 취소되거나 실행 실해하면 하위 자식들은 모두 취소된다.

  -자식의 실패는 부모에 전달되며 부모 또한 실패하고, 다른 모든 자식들도 취소됨

  • SupervisorJob

    자식의 실패가 그 부모나 다른 자식에 전달되지 않으므로 실행을 유지할 수 있음

• join()

  :명시적으로 코루틴이 완료되길 기다림

    import kotlinx.coroutines.*
  
    fun main(){
    	runBlocking{
    	    val job = GlobalScope.launch { //새로운 코루틴 백그라운드에서 실행 
    	        delay(1000L)//1초동안 넌블로킹 지연 -> main은 그대로 쭉 실행 
    	        pritnln("World!")//2초 지연 되는 동안 출력 2  
    	    }
    	    println("Hello ")//먼저 출력한 후 1
    	    //Thread.sleep(2000L)//아직 끝나지 않은 코루틴을 기다리기 위해 2초 지연 
    			job.join() 
    	}
    }
  

  위의 코드는 join을 사용해서 Thread.sleep을 쓴것과 같은 효과가 나타남! (runBlocking으로 감싸줘야함)

  ⇒ world를 출력하는 launch가 완료될때까지 상위루틴인 main을 블록킹 하기 때문에

  • runBlocking : 새로운 코루틴을 실행하고 완료되기 전까지는 현재 스레드를 블로킹

코루틴의 중단과 취소

• 중단(코루틴 코드 내에서)

  -delay(시간값) : 일정 시간을 지연(넌블로킹)하며 중단

  -yield() : 특정 값을 산출하기 위해 중단

• 취소(코루틴 외부)

  -Job.cancel() : 지정된 코루틴 작업을 즉시 취소

  -Job.cancelAndJoin() : 지정된 코루틴 작업을 취소 ( 완료시까지 기다림)

예제

import kotlinx.coroutines.*

suspend fun doWork1() : String {
    delay(1000) //1초 지연 
    return "Work1"
}

suspend fun doWork2() : String {
    delay(3000) //3초 지연 
    return "Work2"
}

private fun workInSerial() : Job {
    val job = GlobalScope.launch {
        val one = doWork1() //1초 지연 후 Work1을 리턴받음 
        val two = doWork2() //3초 지연 후 Work2를 리턴받음 
        println("Kotlin One : $one") 
        println("Kotlin Two : $two")
        //1+3 = 4 초가 지난뒤 one, two 출력 (suspend function은 순차적으로 실행하기 때문) 
    }
    return job
}

fun main() = runBlocking{
		val time = measureTimeMillis {
    val job = worksInserial()
    job.join() //지연되는동안 main이 꺼지지 않게 join사용 
		}
		println("time : $time ") //4초가 나오는걸 알 수 있음 
}

• 특정 suspend 함수를 비동기로 처리하는 방법 !

→ async를 사용하자

import kotlinx.coroutines.*

suspend fun doWork1() : String {
    delay(1000) //1초 지연 
    return "Work1"
}

suspend fun doWork2() : String {
    delay(3000) //3초 지연 
    return "Work2"
}

private fun worksInParallel() : Job { //비동기적 실행 
	val one = GlobalScope.async { //1
			doWork1()
	}
	val two = GlobalScope.async { //2
			doWork2()
	}
	
	val job = GlobalScope.launch { //결과로 반환되는 Deffered<T>를 확인하기 위해 
			//어떤 루틴이 먼저 종료될지 알기 어렵기 때문에, 
			//await을 사용해서 현재 스레드의 블로킹 없이 먼저 종료되면 결과를 가져올 수 있음
			val combined = one.await() + "_" + two.await() 
			println("Kotlin Combined : $combined")
	}
	return job
}

fun main() = runBlocking{
		val time = measureTimeMillis {
    val job = worksInParallel()
    job.join() //지연되는동안 main이 꺼지지 않게 join사용 
		}
		println("time : $time ") //3초가 나오는걸 알 수 있음 
}

async를 사용하면 비동기적으로 다른 코루틴을 수행할 수 있다!

1과 2는 동시에 수행되고, 지연된 결과 값을 받기 위해 await()을 사용 한다.

코루틴 문맥(Coroutine Context)

: 코루틴을 실행하기 위한 다양한 설정값을 가진 관리 정보

ex) 코루틴 이름, 디스패처, 작업 상세사항, 예외 핸들러 등

→ 디스패처는 코루틴 문맥을 보고 어떤 스레드에서 실행되고 있는지 식별 가능

• CoroutineName

  디버깅을 위해서 사용됨

    val someCoroutineName = CoroutineName("someCoroutineName")
  

• Job

  작업 객체를 지정할 수 있으며 취소 가능 여부에 따라 SupervisorJob()사용

    val parentJob = SupervisorJob() // or Job()
    val someJob = Job(parentJob)
  

• CoroutineDispatcher

  Dispatchers.Default, … IO 등을 지정할 수 있으며 필요에 따라 스레드 풀 생성

• CoroutineExceptionHandler

  코루틴 문맥을 위한 예외처리 담당, 코루틴에서 예외가 던져지면 처리

  예외가 발생한 코루틴은 상위 코루틴에 전달되어 처리될 수 있다.

  만일 예외처리가 자식에만 있고 부모에 없는 경우, 부모에도 예외가 전달 됨 → 앱이 crash되기 때문에 주의 !

  예외가 다중으로 발생하면, 최초 하나만 처리하고 나머지는 무시됨

** 코루틴 스코프**

• GlobalScope

  독립형 코루틴 구성, 생명주기는 프로그램 전체에 해당하는 범위를 가지며 main의 생명주기가 끝나면 같이 종료됨

  프로그램 어디서나 제어, 동작이 가능한 기본 범위

  Dispatchers.Unconfined와 함께, 작업이 서로 무관한 전역 범위 실행

  보통 여기에서는 launch나 async 사용이 권장되지 않음

    val scope1 = GlobalScope
    scope1.launch {...}
    scope1.async {...}
  

  or

    GlobalScope.launch{...}
    val job1 = GlobalScope.launch{...} // Job 객체 -> Job.join()으로 기다림
    val job2 = GlobalScope.async{...} // Deffered 객체 -> Deffered.await()으로 기다림 (결과 반환) 
  

• CoroutineScope

  특정 목적의 디스패처를 지정한 범위를 블록으로 구성할 수 있음

  모든 코루틴 빌더는 CoroutineScope의 인스턴스를 갖는다.

  launch와 같이 인자가 없는 경우, CoroutineScope에서 상위의 문맥이 상속되어 결정

  launch(Dispatchers.옵션인자) {…} 와 같이 디스패처의 스케줄러를 지정 가능

  Dispatcher

  • Dispatchers.Default : 기본적인 백그라운드동작
  • Dispatchers.IO : I/O에 최적화 된 동작
  • Dispatchers.Main : 메인(UI) 스레드에서 동작

    val scope2 = CoroutineScope(Dispatchers.Default)
    val routine1 = scope2.launch{...}
    val routine2 = scope2.launch{..}
  

  or

    launch(Dispatchers.Default){...}
    async(Dispatchers.Default){...} 
  

• 예제 #1

    import kotlinx.coroutines.*
  
    fun main()  = runBlocking {
        launch {
            delay(200L)
            println("Task from runBlocking")// #1
        }
  
        coroutineScope {
            launch {
                delay(500L)
                println("Task from nested launch")// #2
            }
            delay(100L)
            println("Task from coroutineScope")// #3
        }
  
        println("end of run Blocking")// #4 
    }
  

  이해하기 어려워서 여기저기 찾아봤고, delay를 다르게 주면서 실행시켰더니 어느정도 이해가 되었다.

  • runBlocking : 자식 스레드가 완료될 때 까지 현재 스레드를 block
  • coroutineScope : 자식 스레드가 완료될 때 까지 현재 스레드를 block 하지 않음

  → main은 runBlocking으로 선언되었기 때문에 자식 스레드인 coroutineScope가 완료될때까지 block되어 있음 (= runBlocking에 선언되어있는 println(“end of run Blocking”) block)

  #1은 delay를 사용해서 non-blocking 되어 0.2초동안 대기하는 동시에 바로 밑의 코드가 실행된다.

  #2는 non-blocking 으로 0.5초동안 대기하기 때문에, 바로 밑의 코드가 실행됨에 동시에 0.1초 동안 대기하는 #3가 제일 빨리 끝나 println(“Task from coroutineScope”)가 실행된다.

  그 다음으로 짧았던 println(“Task from runBlocking”)가 실행되고, 가장 길었던 println(“Task from nested launch”) 가 실행된다.

  마지막으로 자식 스레드인 coroutineScope가 끝났으니, #4가 Blocking에서 해제되고 println(“end of run Blocking”)가 실행된다.

  실행결과

  

• 예제 #2

import kotlinx.coroutines.*

fun main()  = runBlocking {
    launch {
        delay(200L)
        println("Task from runBlocking")// #1
    }

    coroutineScope {
        launch {
            delay(100L)
            println("Task from nested launch")// #2
        }
        delay(100L)
        println("Task from coroutineScope")// #3

    }

    println("end of run Blocking")// #4
}

→ 예제#2는 예제#1에서 delay를 각각 다르게 걸어봤다.

#1에서 delay에 non-blocking으로 0.2초를 대기하는 동시에 바로 밑의 코드인 #2와 #3이 0.1초씩 대기한후 각각 결과를 수행한다. (같은 100 대기를 줘도 둘중 빨리 끝나는 작업먼저 출력)

#1이 0.2초 대기하는 동안 main의 자식 스코프인 coroutineScope가 끝나서 #4가 Blocking에서 해제되고 println(“end of run Blocking”)가 실행된다.

그리고 마지막으로 가장 오래 대기했던 #1 println(“Task from runBlocking”)이 실행된다.

• 스레드풀(thread pool)

: 보통 CommonPool이 지정되어 코루틴이 사용할 스레드의 공동 풀을 사용

이미 초기화 되어있는 스레드 중 하나 혹은 그 이상이 선택되며 초기화 하기 때문에 스레드를 생성하는 오버헤드가 없어 빠르다.

하나의 스레드에 다수의 코루틴이 동작할 수 있음

• 예제 #3

import kotlinx.coroutines.*;

fun main() = runBlocking { //runBlocking -> 자식 코루틴(val request = launch)이 끝날때까지 상위 코루틴 block
    println("runBlocking: $coroutineContext")
    val request = launch{ //#0 request내부에서도, 자식 코루틴(#2)이 끝날때까지 block..?(runBlocking에서 상속받은 대로?)
        println("request: $coroutineContext")
        //#1과 #2는 val request = launch 블록 안의 launch기때문에 자식 코루틴이 된다. 하지만 #1은 GlobalScope기 때문에 부모에 영향을 미치지 않고 독립적이다.
        GlobalScope.launch {//#1 GlobalScope는 부모와 상관없이 프로그램 전역으로 독립적인 수행
            println("job1: before suspend function, $coroutineContext")
            delay(1000)
            println("job1: after suspend function, $coroutineContext")//#4는 부모코루틴과는 개별적이기 때문에 delay(1000)지나고 이 문장 수행
        }
        launch {//#2 부모(#1)의 문맥을 상속 = 상위 launch의 자식, 부모도 Blocking에서 돌고 있기 때문에 자식도 Blocking도 돈다
            //launch(Dispatchers.Default){ //부모의 문맥을 상속 (상위 launch의 자식), Dispatcher가 다르기 때문에 분리된 작업
            //CoroutineScope(Dispatchers.Default).launch {//새로운 스코프가 구성되어 request와 무관
            delay(100)
            println("job2: before suspend function, $coroutineContext")
            delay(1000)
            println("job2: after suspend function, $coroutineContext") //#5 request(부모)가 취소되면 수행되지 않음
        }
    }
    println("??0")
    delay(500) //#3 #2까지 수행을 하고, delay(1000) 전에 여기로 넘어온다.
    println("??1")
    request.cancel()//부모 코루틴(request)의 취소
    println("??2")
    delay(1000)
}

• 실행방법 비교

#1

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

fun work(i : Int){
    Thread.sleep(100)
    println("Work $i done")
}

fun main(){
    val time = measureTimeMillis {
        **runBlocking** { //#1 
            for(i in 1..2){ 
                **launch** { //#2 
                    work(i)
                }
            }
        }
    }
    println("Done in $time ms")
}

: #1의 runBlocking 때문에 launch의 내용이 전부 다 수행될때까지 for문을 block → 작업 2번 각각 단일 스레드에서 순차적으로 실행

#2

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

fun work(i : Int){
    Thread.sleep(100)
    println("Work $i done")
}

fun main(){
    val time = measureTimeMillis {
        **runBlocking** {
            for(i in 1..2){
                **launch(Dispatchers.Default)** {
                    work(i)
                }
            }
        }
    }
    println("Done in $time ms")
}

: Dispatchers.Default 때문에 생성된 2개의 작업은 동시성을 제공하여 분리된 작업으로 실행됨

#3

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

fun work(i : Int){
    Thread.sleep(100)
    println("Work $i done")
}

fun main(){
    val time = measureTimeMillis {
        **runBlocking** {
            for(i in 1..2){
                **GlobalScope.launch** {
                    work(i)
                }
            }
        }
    }
    println("Done in $time ms")
}

: GlobalScope를 붙이면 runBlocking과는 무관한 코루틴 블록이 된다. runBlocking에서는 자식 스레드가 아니라서 기다려주지않고 바로 종료하기 때문에 실행되지 않음 → 직접 job객체를 받아 join()시켜서 기다리게 만들어야함

시작 시점에 대한 속성

launch의 원형

public fun lunch(
	context: CoroutineContext, //문맥
	start : **CoroutineStart**, // 코루틴을 어떤 방식으로 start 할것인가 
	parent : Job?, //부모-자식간의 관계 지정
	onCompletion: CompletionHandler?, //완료됐을때 처리 핸들러 지정 
	block : suspend CoroutineScope.() -> Unit): Job {
	...
}

• CoroutineStart

-DEFAULT : 즉시 시작(해당 문맥에 따라 즉시 스케줄링됨)

-LAZY : 코루틴을 느리게 시작(처음에는 중단된 상태, start()나 async-await()등으로 시작)

val someJob = launch(start = CoroutineStart.LAZY) {
	...
}

Button.setOnClickListener {
	someJob.start()
}

-ATOMIC : DEFAULT와 비슷하나, 코루틴을 실행 전에는 취소할 수 없음 (일단 한 번은 시작되어야하고, 시작 되어야 취소 가능)

-UNDISPATCHED: 현재 스레드에서 즉시 시작 (첫 지연함수(suspend)까지, 이후 재개시 디스패치됨)

runBlocking

: 새로운 코루틴을 실행하고 완료되기 전까지는 현재(caller)스레드를 블로킹

코루틴 빌더와 마찬가지로 CoroutineScope의 인스턴스를 가짐

fun <T> runBlocking(
	context: CoroutineContext = EmptyCoroutineContext, //상위 문맥이 상속됨 
	block: suspend CoroutineScope.() -> T
): T(source) 

• main()블로킹

import kotlinx.coroutines.*

fun main() = runBlocking {
	launch {
		delay(1000L)
		println("World!")
	}
	println("Hello")
}

위에서 봤던 예제와 동일하다!

main에서 runBlocking을 사용했기 때문에 자식 코루틴인 launch가 끝날때까지 main이 block된다.

그리고 launch()안에 아무것도 안써져있기 때문에 runBlocking의 문맥을 그대로 상속받는다.

• 특정 디스패처 옵션

runBlocking(Dispatchers.IO)
	launch{
		repeat(5){
			println("counting ${it + 1}")
			delay(1000)
		}
	}
}

IO → 입출력에 적합한 스레드 옵션으로 설정됨

witContext()

: 인자로 코루틴 문맥을 지정하며, 해당 문맥에 따라 코드 블록을 실행

해당 코드 블록은 다른 스레드에서 수행되며 결과를 반환

부모 스레드는 블록하지 않음 (runBlocking과의 차이점 !)

suspend fun <T> withContext(
		context : CoroutineContext,
		block : Suspend CoroutineScope.() -> T
): T (source)

• withContext(NonCancellable) {…}

: 해당 job이 cancel()요청을 받아 취소되더라도 {…} 의 내용은 꼭 실행을 보장

Scope Builder

• coroutineScope 빌더

  자신만의 코루틴 스코프를 선언하고 생성할 수 있음

  모든 자식이 완료되기 전까지는 생성된 코루틴 스코프는 종료되지 않음 (runBlocking과 유사하지만, runBlocking은 단순 함수로 현재 스레드를 블로킹, coroutineScope는 단순히 지연 함수 형태로 넌블로킹으로 사용됨)

  만약 자식 코루틴이 실패하면, 해당 스코프도 실패하고 남은 모든 자식은 취소됨 ! (반면 supervisorScope는 실패하지 않음)

• supervisorScope 빌더

  코루틴 스코프를 생성하며 이때 SupervisorJob과 함께 생성하여 기존 문맥의 Job을 오버라이드함

  -launch를 사용해 생성한 작업의 실패 : CoroutineExceptionHandler를 통해 핸들링

  -async를 사용해 생성한 작업의 실패 : Deffered.await의 결과에 따라 핸들링

  -parent를 통해 부모작업이 지정되면 자식작업이 되며, 이때 부모에 따라 취소여부 결정

  → 자식이 실패하더라도 이 스코프는 영향을 받지 않으므로 실패하지 않음 ⇒ 예외처리 핸들링 중요

부모와 자식 코루틴과의 관계

• 병렬분해

suspend fun loadAndCombine(name1: String, name2: String): Image {
	val deffered1 = async { loadImage(name1) }
	val deffered2 = async { loadImage(name2) }
	return combineImages(deffered1.await(), deferred2.await())
}

각각 다른 이미지 1, 2 를 가져와서 결합하는 예제다. 여기서 둘중하나가 네트워크 문제로 인해 취소되어도 다른 하나는 계속해서 작업을 이어나갈것이다! → 다른 작업이 취소되면 나머지 작업도 취소되도록 만들자 ⇒ coroutineScope로 묶어주자 !

suspend fun loadAndCombine(name1: String, name2: String): Image = 
coroutineScope {
	val deffered1 = async { loadImage(name1) }
	val deffered2 = async { loadImage(name2) }
	return combineImages(deffered1.await(), deferred2.await())
}

• 스코프 취소 처리

val scope = CoroutineScope
val routine1 = scope.launch {...}
val routine2 = scope.async {...}

scope.cancel()
//scope.cancleChildren() 

try { 
...
} catch ( e: CancellationException ) {
//취소 예외처리
}

스코프가 취소 됐다면, try-catch문을 통해 예외처리를 해주면 된다.

• 실행 시간 제한

  -withTimeout(시간값) : 특정 시간값 동안만 수행하고 블록을 끝냄 → 시간값이 되면 TimeoutCancellationException 예외를 발생

  -withTimeoutOrNull(시간값) : 동작은 위와 동일하지만 → 예외를 발생하지 않고 null 반환

---

참고

https://www.boostcourse.org/mo132/joinLectures/28611

https://eso0609.tistory.com/82

https://tourspace.tistory.com/150

[https://www.youtube.com/watch?v=14AGUuh8Bp8&list=PLbJr8hAHHCP5N6Lsot8SAnC28SoxwAU5A&index=2]