[Kotlin] join, joinAll을 통한 코루틴 순차 처리

launch, async 같은 코루틴 빌더 함수로 코루틴을 만들고 이 안에서 비동기 로직을 처리한 경험이 있을 것이다.

그러다 보면 특정 시점에 작업이 완료될 때까지 기다려야 하는 상황이 생길 수 있다. 여러 API를 동시 호출한 후 모든 결과가 모여야 다음 단계로 넘어갈 수 있는 경우가 그렇다.

이 때 사용할 수 있는 건 코루틴이 제공하는 join, joinAll 함수를 쓰는 것이다. 쓰레드의 join()을 써 봤다면 이해가 쉬울 수 있다.

 

https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-job/join.html

join | kotlinx.coroutines – Kotlin Programming Language

이 작업이 끝날 때까지 코루틴을 일시 정지한다. 호출된 코루틴의 작업이 여전히 활성 상태라면 어떤 이유든 작업이 완료되면 이 호출은 예외 없이 정상 재개된다. 또한 작업이 여전히 New 상태면 해당 코루틴을 시작한다
작업은 모든 자식 작업이 완료된 때만 완료된 걸로 간주된다. 이 일시 정지 함수는 취소 가능하고 호출하는 코루틴의 작업 취소 여부를 항상 확인한다. 이 함수 호출 시점이나 일시 정지 중에 호출 코루틴의 작업이 취소, 완료되면 CancellationException을 발생시킨다
특히 자식 코루틴이 실패했을 경우 부모 코루틴이 자식 코루틴에 대해 join을 호출하면 CancellationException이 발생한다. 이건 자식 코루틴의 실패가 기본적으로 부모를 취소하기 때문이며 자식이 supervisorScope 안에서 실행된 경우는 예외다
이 함수는 onJoin 절과 같이 select 호출에서 쓸 수 있다. 대기 없이 이 작업의 완료 여부를 확인하려면 isCompleted를 써라
cancel, join 호출을 결합한 cancelAndJoin 함수가 있다

 

쓰레드의 join도 코루틴의 join처럼 호출 지점에서 대상의 작업이 완료될 때까지 기다리는 건 동일하지만 쓰레드는 현재 쓰레드를 블로킹하기 때문에 기다리는 동안 쓰레드는 아무것도 못하고 대기 상태로 묶여 있다. OS 수준의 쓰레드 블로킹이 발생하는 건 덤이다.

반면 코루틴의 join은 코루틴만 일시 정지하고 기다리는 동안 쓰레드는 다른 코루틴을 실행할 수 있다. 이 이상의 내용은 이 포스팅의 범위를 넘어서기 때문에 생략한다.

 

아래는 코루틴의 join을 사용하는 예시다.

 

fun main() = runBlocking {
    val job1 = launch {
        delay(1000L)
    }

    val job2 = launch {
        delay(500L)
    }

    job1.join()
    job2.join()
}

 

이걸 실행하면 어떤 식으로 작동하는가?

먼저 launch 블록은 생성되자마자 실행되긴 하지만 각각 1초, 0.5초 딜레이가 있어서 즉시 실행되지 않는다. 그럼 job1.join 위의 코루틴 시작이 먼저 출력될 것이다.

이어서 0.5초로 딜레이가 가장 짧은 job2가 실행된다. 그리고 1초 딜레이를 가진 job1이 실행된다. 이후 job1.join이 job2.join보다 위에 있기 때문에 job1.join, job2.join 순서로 실행된 후 프로그램이 종료된다.

join은 코루틴 실행 순서를 제어하는 게 아닌 완료를 기다리면서 완료 여부를 확인하는 곳임에 주의한다.

 

그럼 아래 코드는 옳게 작동하는 코드인가?

 

fun main() = runBlocking {
    val downloadJob = launch {
        println("다운로드 시작")
        delay(1000L)
        println("다운로드 끝")
    }

    val fileProcessingJob = launch {
        println("파일 처리 시작")
        delay(500L)
        println("파일 처리 끝")
    }

    downloadJob.join()
    fileProcessingJob.join()
    
    println("모든 작업 끝")
}

 

실행해 보기 전에 어떻게 작동할지 생각해 본다.

이 코드를 실행하면 아래처럼 작동한다.

 

 

다운로드와 동시에 파일 처리를 시작하는 걸 볼 수 있다. 실제로 이렇게 작동하는 건 말이 안 된다.

보통 옳은 과정이라면 다운로드가 완료된 후에 파일 처리를 진행하고, 파일 처리가 완료된 후에 모든 작업이 끝났다는 로그가 표시돼야 할 것이다. 이렇게 작동하려면 아래처럼 위치를 수정한다.

 

fun main() = runBlocking {
    val downloadJob = launch {
        println("다운로드 시작")
        delay(1000L)
        println("다운로드 끝")
    }

    downloadJob.join()

    val fileProcessingJob = launch {
        println("파일 처리 시작")
        delay(500L)
        println("파일 처리 끝")
    }

    fileProcessingJob.join()
    println("모든 작업 끝")
}

 

 

downloadJob.join()을 위로 올려야 다운로드 후에 파일 처리가 진행되는 걸 볼 수 있다.

downloadJob이 완료될 때까지 runBlocking 코루틴을 일시 중단해서 downloadJob 내부 코드가 모두 실행되면 runBlocking이 재개되어 fileProcessJob을 이어서 실행한다. 비슷한 경우라면 토큰 업데이트 후 API를 호출해야 하는 경우가 해당될 것이다.

 

다음은 joinAll()을 사용하는 예시다.

 

fun main() = runBlocking {
    val downloadJob = launch {
        delay(1000L)
    }

    downloadJob.join()

    val fileProcessingJob = launch {
        delay(500L)
    }

    fileProcessingJob.join()

    println("=== joinAll() 예시 ===")

    val job1 = launch {
        delay(1000L)
        println("Job 1 완료")
    }

    val job2 = launch {
        delay(500L)
        println("Job 2 완료")
    }

    val job3 = launch {
        delay(1500L)
        println("Job 3 완료")
    }

    // 모든 Job을 리스트로 모아서 한 번에 join -> 가장 긴 Job 완료(1500ms)까지 대기
    joinAll(job1, job2, job3)

    println("모든 Job 완료!")
}

 

joinAll()은 Job 리스트를 받아서 그 Job들이 완료될 때까지 대기하는 함수다.

여러 파일을 모두 다운로드해서 완료까지 기다린 후 그 파일들에 전체적인 처리를 가할 때 다운로드를 기다리기 위해, 또는 이미지들에 어떤 처리를 한 후 모두 변환이 완료되면 서버에 업로드하는 경우 등에 사용할 수 있겠다.

joinAll()의 내부 구현은 아래와 같다.

 

public suspend fun joinAll(vararg jobs: Job): Unit = jobs.forEach { it.join() }