코루틴 빌더와 Job

해당 내용은 코틀린 코루틴의 정석의 5장 내용을 공부하며 정리한 내용입니다.

코루틴 빌더

코루틴 빌더 함수는 코루틴을 만들고 코루틴을 추상화한 Job 객체를 생성한다

우리가 흔히 사용하던 launch 함수 또한 코루틴 빌더이다.

그렇다면 launch는 무엇을 반환할가? → Job 객체를 생성하고 반환한다.

생성된 Job 객체는 코루틴의 상태를 추적 하고 제어 하는 데 사용된다.

fun main() = runBlocking<Unit> {
    val job: Job = launch { 
        println("launch1: ${Thread.currentThread().name}")
    }
}

코루틴은 일시 중단할 수 있는 작업으로 실행 도중 일시 중단된 후 나중에 이어서 실행될 수 있다(중단 및 재개).

코루틴을 추상화한 Job 객체는 이에 대응해 코루틴을 제어할 수 있는 함수와 상태를 나타내는 상태 값들을 외부에 노출한다.

Job 객체를 사용해 코루틴 간 순차 처리를 하는 방법과 코루틴의 상태를 확인하고 조작하는 방법에 대해 알아본다.

---

join을 사용한 코루틴 순차 처리

순차 처리가 필요한 경우

만약 토큰이 업데이트 완료 되기전에 네트워크 요청을 하면 401(Unauthorized) 코드를 받게된다.

fun main() = runBlocking<Unit> {
    val updateToken: Job = launch(Dispatchers.IO) {
        println("[${Thread.currentThread().name}] Token Update Start")
        delay(100L)
        println("[${Thread.currentThread().name}] Token Update Finish")
    }

    val networkCallJob: Job = launch(Dispatchers.IO) {
        println("[${Thread.currentThread().name}] Network Call Start")
        delay(100L)
        println("[${Thread.currentThread().name}] Network Call Finish")
    }
}
[DefaultDispatcher-worker-1 @coroutine#2] Token Update Start
[DefaultDispatcher-worker-3 @coroutine#3] Network Call Start
[DefaultDispatcher-worker-1 @coroutine#3] Network Call Finish
[DefaultDispatcher-worker-1 @coroutine#2] Token Update Finish

위의 결과를 보면 토큰 업데이트가 완료되기 전에 네트워크 요청을 보내는 모습을 볼 수 있다.

즉, 위 코드에서 토큰 업데이트 작업과 네트워크 요청 작업이 병렬로 동시에 실행되는 문제가 발생!!

위 문제를 해결하려면?

1. 토큰 업데이트 작업을 완료한다.
2. 토큰 업데이트 작업이 완료되면 네트워크 작업을 실행한다

Job 객체를 이런 문제 해결을 위해 순차 처리할 수 있는 join 함수를 제공한다.

fun main() = runBlocking<Unit> {
    val updateToken: Job = launch(Dispatchers.IO) {
        println("[${Thread.currentThread().name}] Token Update Start")
        delay(100L)
        println("[${Thread.currentThread().name}] Token Update Finish")
    }
    updateToken.join()
    val networkCallJob: Job = launch(Dispatchers.IO) {
        println("[${Thread.currentThread().name}] Network Call Start")
        delay(100L)
        println("[${Thread.currentThread().name}] Network Call Finish")
    }
}
[DefaultDispatcher-worker-1 @coroutine#2] Token Update Start
[DefaultDispatcher-worker-1 @coroutine#2] Token Update Finish
[DefaultDispatcher-worker-1 @coroutine#3] Network Call Start
[DefaultDispatcher-worker-1 @coroutine#3] Network Call Finish

첫 번째 코드와 달라진 점은 updateToken.join() 코드가 추가되었다.

join의 대상이 된 코루틴의 작업이 완료될 때까지 join을 호출한 코루틴이 일시 중단된다.

즉, 위 코드에서처럼 runBlocking 코루틴이 updateToken.join() 을 호출하면 runBlocking 코루틴은 updateTokenJob 코루틴이 완료될 때까지 일시 중단된다.

이후 updateToken 내부의 코드가 모두 실행되면 runBlocking 코루틴이 재개돼 networkCallJob 을 실행한다.

이와 같이 우리가 의도한 동작대로 동작시킬 수 있다.

---

joinAll을 사용한 코루틴 순차 처리

빵빵이가 셀카 사진을 SNS에 3장 업로드 한다고 가정해보자.

서버에서 규정한 이미지 크기, 형식을 맞추기 위해 각 사진을 변환(압축 / 포맷 변경 등) 후 이미지를 서버에 업로드 한다.

하나의 이미지 변환기로 3개의 이미지를 순차적으로 처리하는 것 보다

각 이미지 변환 작업을 병렬로 실행한 후 결과를 취합해 업로드 작업을 실행하는 것이 효율적이다.

이런 작업을 위해 복수의 코루틴의 실행이 모두 끝날 때까지 호출부의 코루틴을 일시 중단시키는 joinAll 함수를 제공한다.

// joinAll = 내부적으로 forEach문에 각각의 Job에 join을 걸어주는 로직
public suspend fun joinAll(vararg jobs: Job): Unit = jobs.forEach { it.join() }

fun main() = runBlocking<Unit> {
    val convertImageJob1: Job = launch(Dispatchers.Default) {
        Thread.sleep(1000L)
        println("[${Thread.currentThread().name}] convert image 1")
    }

    val convertImageJob2: Job = launch(Dispatchers.Default) {
        Thread.sleep(1000L)
        println("[${Thread.currentThread().name}] convert image 1")
    }

    joinAll(convertImageJob1, convertImageJob2)
    
    val uploadImageJob: Job = launch(Dispatchers.IO) {
        println("[${Thread.currentThread().name}] upload image")
    }
}
[DefaultDispatcher-worker-1 @coroutine#2] convert image 1
[DefaultDispatcher-worker-2 @coroutine#3] convert image 2
[DefaultDispatcher-worker-1 @coroutine#4] upload image

여기서 delay 대신 Thread.sleep을 하는 이유는 이미지를 변환하는 작업은 CPU 바운드 작업이므로 각 코루틴을 Dispatchers.Default에 실행 요청하며, 내부에서는 sleep을 통해 작업이 진행되는 동안 스레드를 블로킹하게 만든다.

Default 디스패처는 작업을 실행하는 동안 스레드가 블록된다.

joinAll 이 호출되는 순간 ConvertImage1, ConvertImage2가 완료될 때 까지 runBlocking 코루틴은 잠시 일시 중단된다 . 이후 두 작업이 모두 완료되면 다시 재개 되어 uploadImage을 실행 요청한다.

코루틴을 순차적으로 처리하는 방법을 알아보았다

이번에는 Job 객체를 생성해 놓고 내가 원하는 시점에 실행할 수 있도록 만드는 방법에 대해 알아보자.

---

CoroutineStart.LAZY 사용해 코루틴 지연 시작하기

지금까지의 예제를 살펴보면 코루틴 빌더 launch를 사용하면 바로바로 코루틴이 실행되었다.

이번에는 “아 Job 객체 생성해 놓고 나중에 실행시키고 싶은걸?“ 방법에 대해 알아보겠습니다.

예제에 들어가기 앞서 아래의 함수를 사용하겠습니다.(걸린 시간 체크를 위해)

fun getElapsedTime(startTime: Long): String {    
			return "걸린 시간 : ${System.currentTimeMillis() - startTime}ms"
}

launch 함수의 파라미터를 보면

public fun CoroutineScope.launch(
    context: CoroutineContext = EmptyCoroutineContext,
    start: CoroutineStart = CoroutineStart.DEFAULT,
    block: suspend CoroutineScope.() -> Unit
): Job

context, start, block이 있다.

여기서 start 파라미터를 통해 원하는 시점에 코루틴을 시작시킬 수 있다.

기본값으로 CoroutineStart.DEFAULT 로 할당이 되어있으며 이는 즉시 실행함을 의미한다.

start 파라미터에 어떠한 것들이 들어갈 수 있는지 간단하게 알아보겠습니다.

종류	설명	내부 함수
DEFAULT	디스패처에 따라 즉시 실행	startCoroutineCancellable
ATOMIC	즉시 실행하지만, 취소 불가 상태에서 시작됨	startCoroutine
UNDISPATCHED	현재 스레드에서 바로 실행, suspend 지점부터 디스패처로 전환	startCoroutineUndispatched
LAZY	시작 안함, start() 해야 시작함	Unit // will start lazily

코루틴을 먼저 생성해 놓고 나중에 실행해야 하는 상황이 생길 때는 LAZY 를 사용해야 합니다.

fun main() = runBlocking<Unit> {
    val startTime = System.currentTimeMillis()
    launch(start = CoroutineStart.LAZY) {
        println("launch1: ${getElapsedTime(startTime)}")
    }
}
// 아무것도 출력 안됨

위 코드를 실행시켜보면 start를 LAZY 로 했기 때문에 아무것도 출력이 되지 않습니다.

fun main() = runBlocking<Unit> {
    val startTime = System.currentTimeMillis()
    val lazyJob: Job = launch(start = CoroutineStart.LAZY) {
        println("[${Thread.currentThread().name}] [${getElapsedTime(startTime)}]")
    }
    delay(1000L)
    lazyJob.start()
}
[main @coroutine#2] [지난 시간 : 1019ms]

이번에는 runBlocking 코루틴에 delay(1초)를 걸고 lazyJob.start() 를 실행시켜보겠습니다.

1초 뒤 코루틴 start() 가 되었기 때문에 지난시간에는 1019ms 가 나오게 됩니다.

---

코루틴 취소하기

코루틴 실행 도중 코루틴을 실행할 필요가 없어지면 즉시 취소해야 합니다.

ex) 사용자가 파일 변환 작업을 요청했지만 단순 변심으로 취소시킴 → 코루틴 종료시켜야 함.

이 때 코루틴이 계속 실행된다면 스레드를 계속 사용하며 성능 저하 로 이어진다.

이러한 상황을 막기 위해 Job 객체는 코루틴을 취소할 수 있는 cancel 함수를 제공한다.

fun main() = runBlocking<Unit> {
    val startTime = System.currentTimeMillis()
    val longJob: Job = launch(Dispatchers.Default) {
        repeat(10) { repeatTime->
            delay(1000L)
            println("[${getElapsedTime(startTime)}] 반복횟수: $repeatTime")
        }
    }
    delay(3500L)
    longJob.cancel()
}
[지난 시간 : 1025ms] 반복횟수: 0
[지난 시간 : 2046ms] 반복횟수: 1
[지난 시간 : 3051ms] 반복횟수: 2

위 코드를 보면 1초마다 println 하는 코루틴을 생성, 3.5초 뒤에 생성한 코루틴. cancel 을 했기 때문에

3번 출력된 뒤 종료된 모습을 확인할 수 있습니다.

---

cancelAndJoin을 사용한 순차 처리

위 예제에서 코루틴을 취소하는 기능을 구현했다.

fun main() = runBlocking<Unit> {
    val startTime = System.currentTimeMillis()
    val longJob: Job = launch(Dispatchers.Default) {
        repeat(10) { repeatTime->
            delay(1000L)
            println("[${getElapsedTime(startTime)}] 반복횟수: $repeatTime")
        }
    }
    longJob.cancel()
    executeAfterJobCancelled() // 코루틴 취소 후 실행돼야 하는 동작 꼭 취소 후 동작해야 함
}

executeAfterJobCancelled() 함수는 이름에서부터 알 수 있듯이 longJob이 취소된 후 실행되어야 하는 작업입니다.

위 코드는 잘 동작할 것처럼 보이지만 순차성 관점에서 중요한 문제점을 가집니다.

cancel 을 호출하면 코루틴은 즉시 취소되는 것이 아니라 Job 객체 내부의 취소 확인용 플래그를 취소 요청됨 으로 변경함으로써 코루틴이 취소돼야 한다는 것만 알립니다.

이후 미래의 어느 시점에 코루틴의 취소가 요청됐는지 체크하고 취소됩니다.

즉, cancel 시 Job 객체는 곧바로 취소되는 것이 아니라 미래의 어느 시점에 취소된다.

이 순차성 보장을 위해 코루틴은 cancelAndJoin 함수를 제공합니다.

cancelAndJoin 함수를 호출하면 cancelAndJoin 대상이 된 코루틴의 취소가 완료될 때까지 호출부의 코루틴이 일시 중단됩니다.

fun main() = runBlocking<Unit> {
    val startTime = System.currentTimeMillis()
    val longJob: Job = launch(Dispatchers.Default) {
        repeat(10) { repeatTime->
            delay(1000L)
            println("[${getElapsedTime(startTime)}] 반복횟수: $repeatTime")
        }
    }
    longJob.cancelAndJoin()
    executeAfterJobCancelled()
}

executeAfterJobCancelled() 함수가 longJob이 취소 완료된 후에 실행되는 것을 보장할 수 있습니다.

---

코루틴의 취소 확인

위에서 공부했듯이 cancel 함수나, cancelAndJoin 함수를 사용했다고 해서 코루틴이 즉시 취소되는 것이 아니다. Job 객체 내부에 있는 취소 확인용 플래그를 바꾸기만 하며, 코루틴이 이 플래그를 확인하는 시점에 비로소 취소된다.

만약 코루틴이 이 취소를 확인할 수 있는 시점이 없다면 취소는 일어나지 않습니다.

엥? 그럼 코루틴이 취소를 확인하는 시점은 대체 언제일가요?

일시 중단 지점 / 코루틴이 실행을 대기하는 시점

이 시점들이 없다면 코루틴은 취소가 되지 않는다.

fun main() = runBlocking<Unit> {
    val whileJob: Job = launch(Dispatchers.Default) {
        while(true) {
            println("작업 중...")
        }
    }
    delay(100L) // 0.1초 뒤
    whileJob.cancel() // 종료시킴
}
작업 중...
...
... 
무한 작업 중...

cancel 을 요청했지만 작업 중… 이 무한으로 실행된다.

왜그럴까?

코루틴은 일반적으로 실행 대기 시점 / 일시 중단 지점 에 취소를 확인한 후 취소되는데 whileJob 코루틴은 while문에서 코드가 반복되고 있어 while문을 벗어날 수 없고 cancel 을 했지만 코루틴의 취소를 확인할 수 있는 시점이 없기 때문에 계속 실행된다.

이 코드가 취소되도록 만드는 데는 세 가지 방법이 있습니다.

delay를 사용한 취소 확인

delay 함수는 이리 중단 함수(suspend fun)이기 때문에 특정 시간만큼 호출부의 코루틴을
일시 중단하게 만든다.

코루틴은 일시 중단되는 시점에 코루틴의 취소를 확인하기 때문에 작업 중간에 delay(1L) 을 주게 되면 while 문이 반복될 때마다 1밀리초만큼 일시 중단 후 최소를 확인할 수 있습니다.

fun main() = runBlocking<Unit> {
    val whileJob: Job = launch(Dispatchers.Default) {
        while(true) {
            delay(1L)
            println("작업 중...")
        }
    }
    delay(100L) // 0.1초 뒤
    whileJob.cancel() // 종료시킴
}
작업 중...
작업 중...

Process finished with exit code 0

음… 그럼 while문이 돌때마다 1밀리초를 대기해야 하네?…

때문에 이 점에서 효율적이지 않고 성능 저하가 일어날 수 있습니다.

yield를 사용한 취소 확인

yield 는 양보라는 뜻으로 yield 함수가 호출되면 코루틴은 자신이 사용하던 스레드를 양보한다.

스레드를 양보한다는 것은 스레드 사용을 중단한다는 뜻이므로 yield를 호출한 코루틴이 일시 중단되며 이 시점에 취소 됐는지 체크가 일어납니다.

fun main() = runBlocking<Unit> {
    val whileJob: Job = launch(Dispatchers.Default) {
        while(true) {
            println("작업 중...")
            yield()
        }
    }
    delay(100L) // 0.1초 뒤
    whileJob.cancel() // 종료시킴
}
작업 중...
작업 중...

Process finished with exit code 0

“작업 중…”이 출력될 때마다 yield 로 인해 일시 중단이 일어납니다.

일시 중단 후 곧바로 재개되지만 잠깐 일시 중단된 시점에 취소 체크가 일어난다.

하지만 yield를 사용하는 방법 또한 while문을 한 번 돌 때마다 스레드 사용이 양보되면서 일시 중단되는 문제가 있습니다. 매번 일시 중단되는 것은 비효율적입니다.

CoroutineScope.isActive를 사용한 취소 확인!!

CoroutineScope는 코루틴이 활성화됐는지 확인할 수 있는 Boolean 타입의 프로퍼티인 isActive 를 제공한다. 코루틴에 취소가 요청되면 isActive 프로퍼티의 값은 false로 바뀐다.

fun main() = runBlocking<Unit> {
    val whileJob: Job = launch(Dispatchers.Default) {
        while(this.isActive) {
            println("작업 중...")
        }
    }
    delay(100L) // 0.1초 뒤
    whileJob.cancel() // 종료시킴
}
작업 중...
작업 중...

Process finished with exit code 0

이 방법을 사용하면 코루틴이 잠시 멈추지도 않고 스레드 사용을 양보하지도 않으면서 계속해서 작업을 할 수 있어서 효율적이다.

---

코루틴의 상태와 Job의 상태 변수

• 생성: 코루틴 빌더를 통해 코루틴을 생성하면 코루틴은 생성 상태에 놓이며, 자동으로 실행 중 상태로 넘어간다. 만약 실행 중을 늦추고 싶다면 코루틴 빌더의 start 인자로 CoroutineStart.Lazy를 넘기면 된다.

• 실행 중: 지연 코루틴이 아닌 코루틴을 만들면 자동으로 실행 중 상태로 바뀐다. 코루틴이 실제로 실행 중일 때뿐만 아니라 실행된 후에 일시 중단된 때도 실행 중 상태로 본다.

• 실행 완료: 코루틴의 모든 코드가 실행 완료된 경우 실행 완료 상태

• 취소 중: Job.cancel() 등을 통해 취소가 요청됐을 경우 취소 중 상태로 넘어가며, 이는 아직 취소된 상태가 아니어서 코루틴은 계속해서 실행된다.

• 취소 완료: 코루틴의 취소 확인 시점(일시 중단 등)에 취소가 확인된 경우 취소 완료 상태가 된다. 이 상태에서는 코루틴은 더 이상 실행되지 않는다.

Job 객체에서 외부로 공개하는 코루틴의 상태 변수는 isActive , isCancelled, isCompleted 세 가지이며, 각 변수 모두 Boolean 타입 변수이다.

• isActive : 코루틴이 활성화돼 있는지의 여부. 활성화돼 있으면 true 아니라면 false 반환. 활성화돼 있다는 것은 코루틴이 실행된 후 취소가 요청되거나 실행이 완료되지 않은 상태라는 의미. 따라서 취소가 요청되거나 실행이 완료된 코루틴은 활성화되지 않은 것으로 본다.

• isCancelled : 코루틴이 취소 요청됐는지의 여부. 취소 요청되면 true를 반환. 요청되기만 하면 true가 반환되므로 isCancelled가 true이더라도 즉시 취소되는 것은 아니다.

• isCompleted : 코루틴 실행이 완료됐는지의 여부. 모든 코드가 실행완료되거나 취소 완료되면 true를 반환하며, 실행 중인 상태에서는 false를 반환한다.

생성 상태의 코루틴

fun main() = runBlocking<Unit> {
   val job: Job = launch(start = CoroutineStart.LAZY) {
       delay(1000L)
   }
    printJobState(job)
}
Job State
isActive >> false
isCancelled >> false
isCompleted >> false

launch의 start를 CoroutineStart.LAZY 로 설정했기 때문에 실행 중으로 바로 넘어가지 않는다.

때문에 job은 생성 상태에 놓이게 된다.

실행 중 상태의 코루틴

fun main() = runBlocking<Unit> {
   val job: Job = launch {
       delay(1000L)
   }
    printJobState(job)
}
Job State
isActive >> true
isCancelled >> false
isCompleted >> false

start에 LAZY를 삭제했기 때문에 바로 실행 중으로 넘어가게 된다. isActive 만 true인 것을 확인할 수 있다.

실행 완료 상태의 코루틴

fun main() = runBlocking<Unit> {
   val job: Job = launch {
       delay(1000L)
   }
    delay(2000L)
    printJobState(job)
}
Job State
isActive >> false
isCancelled >> false
isCompleted >> true

job은 1초뒤 모든 코드가 다 실행되었기 때문에 종료된다.

그렇기 때문에 2초후 job의 상태를 찍었을 때 isCompleted 가 true인 것을 확인할 수 있다.

취소 중인 코루틴

앞선 내용에서 취소 요청이 되기만해도 isCancelled 는 truer가 된다.

fun main() = runBlocking<Unit> {
   val whileJob: Job = launch(Dispatchers.Default) {
       while(this.isActive) {
           println("작업 중")
       }
   }
    whileJob.cancel()
    printJobState(whileJob)
}
작업 중
작업 중
Job State
isActive >> false
isCancelled >> true
isCompleted >> false

Process finished with exit code 0

위의 코드를 보자. 위에서 공부했던 cancelAndJoin 의 필요성을 느낄 수 있다.

cancel 을 통해 코루틴을 취소한 후 whileJob의 상태를 출력해보았다.

isCancelled 는 true이다 왜? 취소 요청이 왔으니간

하지만 isCompleted 는 false가 나온 것을 확인할 수 있다.

그렇다면 cancelAndJoin() 을 사용해보자.

fun main() = runBlocking<Unit> {
   val whileJob: Job = launch(Dispatchers.Default) {
       while(this.isActive) {
           println("작업 중")
       }
   }
    whileJob.cancelAndJoin()
    printJobState(whileJob)
}
작업 중
작업 중
Job State
isActive >> false
isCancelled >> true
isCompleted >> true

Process finished with exit code 0

위의 결과와 달리 isCompleted 또한 true가 된 것을 확인할 수 있습니다.

isCancelled 또한 취소 요청이 왓었기 때문에 true를 반환합니다.

---

마무리

코루틴 내부에서 어떤 상태 전이가 일어나는지 제대로 아는 것은 매우 중요하다.

따라서 코루틴의 내부 상태가 어떤지에 대해 계속해서 생각하는 습관을 기르자!!