Kotlin Coroutine - (3) Coroutine lifeCycle And Error Handling

비동기 함수를 아래와 같은 두그룹으로 나눌 수 있다.

- 결과가 없는 비동기 함수 : 로그 작업이나, 분석을 위한 데이터 전송 같은 경우, 완료 여부를 확인하지만, 결과가 필요는 없다.

- 결과를 돌려주는 비동기 함수 : 위의 경우 제외 거의 모든 경우... 허허허

위 두가지 경우 모두 예외가 발생하면 대응하거나, 작업이 필요하지 않을경우 취소한다. 따라서 두가지 경우에 대해서 코루틴은 어떻게 하는지 알아보자.

---

Job

잡은 2장에서 말했듯이 fire and forget 작업으로, 결과가 없는 비동기 함수이다. 한번 시작된 작업은 예외가 발생하지 않는한 대기 하지 않는다. 보통 아래와 같이 launch를 통해 Job을 생성한다.

fun asyncFunction() = runBlocking {
    // launch를 통해 Job을 생성.
    val job = GlobalScope.launch {
        ~~~
    }
    // Job() 팩토리 함수로 생성.
    val mJob = Job()
}

 

예외처리

기본적으로 잡 내부에서 발생하는 예외는 잡을 생성한곳 까지 전파된다. 따로 잡이 완료되기를 기다리지 않아도 발생한다.

fun main(args: Array<String>) = runBlocking {
    GlobalScope.launch {
        // Do nothing
    }
    delay(500)
}

이렇게 위와 같은 코드를 실행하면, 현재 스레드의 Uncaught Exception Handler에 예외가 전파된다.

 

라이프 사이클

Job의 라이프 사이클

위 Job의 다이어그램에는 5가지 상태가 있다.

- New : 존재는 하지만 실행되지 않은 잡
- Active : 실행 중인잡. 일시 중단된 잡도 Active로 본다.
- Completed : 잡이 더 이상 실행되지않는 경우.
- Canceling : 실행 중인 잡에서 cancel()이 호출되면 취소가 완료될때까지의 상태.
- Canceled : 취소로 인해 실행이 완료된 잡, 취소된 잡도 완료로 간주된다.

New

잡은 기본적으로 launch()나 Job()을 사용해 생성될때 자동으로 시작된다. 따라서 잡을 만들고 바로 시작되지 않게 하려면 CoroutineStart.Lazy를 사용하면 된다.

// Lazy로 Job을 실행시키지 않아서 예외가 발생하지 않음.
fun main(args: Array<String>) = runBlocking {
    GlobalScope.launch(start = CoroutineStart.LAZY) {
        // Do nothing...
    }
    delay(500)
}

Active

New 상태의 잡은 일반적으로 start()나 join()을 호출해서 실행한다.
start()는 Job이 완료될때까지 기다리지 않고 잡을 시작하는 반면,
join()은 Job이 완료될때까지 실행을 일시 중단한다는 점이다.

fun jobStart() {
    val job = GlobalScope.launch(start = CoroutineStart.LAZY) {
        delay(3000)
    }
    
    job.start()
}

fun jobJoin() = runBlocking {
    val job = GlobalScope.launch(start = CoroutineStart.LAZY) {
        delay(3000)
    }
    
    job.join()
}

여기서 한가지 start()는 실행을 일시 중단하지 않아서, 일시중단 함수나 코루틴에서 실행할 필요가 없다. 어디서든 호출 할 수 있다.
반면에 join()은 일시 중단하므로, 코루틴 또는 일시중단함수에서 실행해야한다.

Active에서 중요한 점은 시작된 모든 잡은 Active상태이고 실행이 완료되거나 취소가 될때까지 Active상태라는 점이다.

Cancelling

cancel()을 통해 Active상태의 Job은 cancelling 상태에 들어갈 수 있다.

fun cancelJob() = runBlocking {
    val job = GlobalScope.launch() {
        //
    }
    //일반적으로 멈출 수 있다.
    job.cancel()
    
    //cause를 통해 원인을 전달해 줄 수 있다.
    job.cancel(cause = CancellationException("timeout!"))
}

cancelAndJoin() 함수도 있다. 이름처럼 실행을 취소하면서 취소가 끝날때 까지 대기한다.

Cancelled

취소되거나 처리되지 않은 예외로 종료된 잡은 cancelled 로 간주된다.
잡이 취소되면, getCancellationException() 함수를 통해 취소에 대한 정보를 받을 수 있다.

fun cancelJob() = runBlocking {
    val job = GlobalScope.launch() {
        //
    }

    //cause를 통해 원인을 전달해 줄 수 있다.
    job.cancel(cause = CancellationException("timeout!"))
    
    val cancellation = job.getCancellationException()
    printf(cancellation.message)
}

취소된 잡과 예외로 인해 실패한 잡을 구별하기 위해 CoroutineExceptionHandler를 설정해 취소 작업을 처리하는 것이 좋다.

fun cancelJob() = runBlocking {
    val exceptionHandler = CoroutineExceptionHandler {
        _: CoroutineContext, throwable: Throwable ->
        println("Job cancelled due to ${throwable.message}")
    }

    GlobalScope.launch(exceptionHandler) {
        //
    }
    
    delay(2000)
}

다음과 같이 invokeOnCompletion()을 사용할 수도 있다.

fun cancelJob() = runBlocking<Unit> {
    GlobalScope.launch() {
        
    }.invokeOnCompletion { cause ->
        cause?.let {
            println("Job cancelled due to ${it.message}")
        }
    }

    dealy(2000)
}

하지만 invokeOnCompletion의 네이밍 처럼, completed된 상황에서도 불려진다 (cause 가 null)
따라서 cause 존재 여부로 잘 종료된것인지, 문제가 있는지 확인할 수있다.

(하지만 개인적으로 Cancel과 Completed가 동시에 처리된다는점에서 좋지 않은듯)

Completed

실행이 중지된 잡은 completed로 간주된다. 이는 정상종료나, 취소됐거나, 예외때문에 종료됐는지 여부에 관계없이 적용된다.

잡의 현재 상태 확인

잡의 상태확인을 위해 3가지 속성을 갖고 있다.

- isActive : 활성상태인지 확인, 일시중지된 경우도 true
- isCompleted : 잡의 실행 완료 여부
- isCancelled : 잡의 취소여부, 취소 요청시 바로 true

상태(State)	isActive	isCompleted	isCancelled
Created	false	false	false
Active	true	false	false
Cancelling	false	false	true
Cancelled	false	true	true
Completed	false	true	false

---

Deferred

디퍼드는 잡과 다르게 결과를 갖는 비동기 작업을 수행한다.
기본적인 컨셉은 연산이 객체를 반환하며, 객체는 비동기 작업이 완료될때까지 비어있다.

디퍼드와 그 상태의 라이프 사이클은 잡과 비슷하다. 차이점은 반환유형과 에러 핸들링 방법이다.

디퍼드는 아래와 같이 async나 CompletableDeferred의 생성자를 활용할 수 있다.

fun main(args: Array<String>) = runBlocking {
    val headlinesTask = GlobalScope.async {
        getHeadlines()
    }
    
    headlinesTask.await()
    
    // CompletableDeferred의 생성자 활용
    val articlesTask = CompletableDeferred<List<Article>>()
}

예외처리

잡과 달리 디퍼드는 처리되지 않은 예외를 자동으로 전파하지 않는다. (디퍼드는 결과를 대기할것으로 예상하기 때문)
따라서 실행시 성공했는지 확인하는건 사용자의 몫이다.

fun delayDeferred = runBlocking {
    val deferred = GlobalScope.async {
    
    }
    // 지연된 실패를 갖지만 예외는 전파가 안됨.    
    delay(2000)
}

fun awaitDeferred = runBlocking<Unit> {
    val deferred = GlobalScope.async {
    
    }
    // 예외가 전파됨.  
    deferred.await()
}

디퍼드의 실행이 코드 흐름의 필수적인 부분임을 나타내는 것이기 때문에 await()을 호출하는 이런 방식으로 설계됐다. 이 방법을 사용하면 명령형으로 보이는 비동기 코드를 보다 쉽게 작성할 수 있고, try-catch 블록을 사용해 예외를 잡을 수 있다.

fun awaitDeferred = runBlocking<Unit> {
    val deferred = GlobalScope.async {
    
    }
    
    try {
        deferred.await()
    } catch (throwable: Throwable) {
        println("Defferred cancelled due to ${throwable.message}")
    }
}

Deferred 또한 Job처럼 CoroutineExceptionHandler를 사용할 수 있다.

---

상태는 한 방향으로만 이동.

일단 잡이 특정 상태에 도달하면 이전 상태로 되돌아 가지 않는다.

fun main(args: Array<String>) = runBlocking {
    val time = measureTimeMillis {
        val job = GlobalScope.launch {
            delay(2000)
        }
        //대기
        job.join()
        
        //재시작
        job.start()
        job.join()
    }
    println("Took $time ms")
}

위 코드는 2초간 실행을 중지하는 잡을 만들고 실행한다. 그리고 한번더 재시작을 위해 Completed된 job을 start()하고 join()을 통해 기다린다.

왠지 코드만 보면 4000ms 이상이 걸릴것 같지만 실제로는 2000ms 정도 걸린다. (한번만 실행됨)

이 이유는 일단 job이 특정 상태에 도달하면 이전으로 돌아가지 않는 다는 특징이 있기 때문이다.

Complete된 잡은 join()을 호출해도 아무 일도 일어나지 않는다. 이미 작업이 완료됐으므로 일시중단도 안일어난다.

따라서 Cancelled와 Completed상태는 final state로 간주된다.

---

RSS- 여러 피드에서 동시에 읽기

2장에서 만들던 RSS리더에 여러피드를 호출하는 기능을 만들어보자.

피드목록지원

여러 피드에서 뉴스를 호출하기 위해 리스트를 하나 만든다.

class MainActivity : AppCompatActivity() {
        
        val feeds = listOf(
            "https://www.npr.org/rss/rss.php?id=1001",
            "http://rss.cnn.com/rss/cnn_topstories.rss",
            "http://feeds.foxnews.com/foxnews/politics?format=xml"
        )
        ...
}

안드로이드 Pie(API28)부터 cleartext HTTP를 비활성화했기 때문에 http접근을 허용하지 않는다.
사용하기 위해서는 manifest에서 android:usesCleartextTraffic="true" 를 해줘야한다.

기존에 만들어둔 fetchRssHeadlines() 함수도 feed를 호출하도록 수정한다.

private fun asyncFetchRssHeadlines(feed: String, 
        dispatcher: CoroutineDispatcher) = GlobalScope.async(dispatcher) {
    val builder = factory.newDocumentBuilder()
    val xml = builder.parse(feed)
    val news = xml.getElementsByTagName("channel").item(0)
    
    (0 until news.childNodes.length)
        .map { news.childNodes.item(it) }
        .filter { Node.ELEMENT_NODE == it.nodeType }
        .map { it as Element }
        .filter { "item" == it.tagName }
        .map {
            it.getElementsByTagName("title").item(0).textContent
        }
}

스레드풀 만들기

크기가 2인 스레드 풀을 만들고 이름을 IO로 한다.

val dispatcher = newFixedThreadPoolContext(2,"IO")

asyncFetchHeadlines()는 서버에서 정보도 가져올 뿐더러 파싱도 하기때문에 풀의 크기를 늘려준다. XML을 파싱하는 오버 헤드는 단일 스레드를 사용하는 경우 성능에 영향을 준다.

데이터를 동시에 가져오기

이제 여러 피드에 요청을 보내기 위한 필요한 모든것을 갖췄다. 목록에서 각 피드당 하나의 디퍼드를 생성한다.
먼저 asyncLoadNews() 함수를 수정해 대기하는 모든 디퍼드를 추적할 수 있는 목록을 만들자.

private fun asyncLoadNews() = GlobalScope.launch {
    //대기하는 모든 deferred를 추적할 수 있는 목록 만들기.
    val requests = mutableListOf<Deferred<List<String>>>()
        
    // 각 피드별로 가져온 요소를 피드 목록에 추가.
    feeds.mapTo(requests) {
        asyncFetchRssHeadlines(it, dispatcher)
    }
        
    // 각 코드가 완료될때까지 대기하는 코드.
    requests.forEach {
        it.await()
    }
}

응답 병합

현재 구현된 asyncLoadNews()는 각 요청이 끝날때까지 대기한다. 하지만 각각 피드마다 헤드라인의 목록을 반환하기 때문에 이들을 하나의 리스트에 담고 싶을 것이다. 이를 위해 각 디퍼드의 내용을 flat map을 이용해 담을 수 있다.

    //각 deferred 내용을 담는 headlines 변수 생성.
    val headlines = requests.flatMap { 
        it.getCompleted()
    }

이후 출력을 위해 ui코드를 개선한다.

    val newsCount = findViewById<TextView>(R.id.newsCount)
    launch(Dispatchers.Main) {
        newsCount.text = "Found ${headlines.size} news in ${requests.size} feeds"
    }

이렇게 따라하면 기본적인 로딩은 될 것이다.

---

예기치 않은 중단.

현재 코루틴이 완료될 때까지 await()를 사용하므로, 코루틴 내부에서 에러가 발생시 현재 스레드로 전파된다. 따라서

- 인터넷이 연결되지 않은 경우
- 하나 이상의 피드 URL이 유효하지 않거나 서버에 문제가 생긴경우

위 두가지 앱이 중단되는 시나리오가 존재한다. 한번 시험해보자.

    //피드에 잘못된 URL 추가
    val feeds = listOf(
        "https://www.npr.org/rss/rss.php?id=1001",
        "http://rss.cnn.com/rss/cnn_topstories.rss",
        "http://feeds.foxnews.com/foxnews/politics?format=xml",
        "htt:myNewsFeed"
    )

위 코드로 바꾸고 실행하면 앱이 종료될것이다.

따라서 예외를 막기위해 대비를 해보자. 제일 쉬운 방법인 await()대신 join()을 써서 예외가 전파되지 않게 한다.

    // 각 코드가 완료될때까지 대기하는 코드.
    requests.forEach {
        it.join()
    }

이후 deferred가 실패하지 않았을때만 getCompleted()가 호출되게 한다. 이 처리를 하지않으면 요청을 읽을때 예외가 전파된다.

    //각 deferred 내용을 담는 headlines 변수 생성.
    val headlines = requests
        .filter { !it.isCancelled }
        .flatMap {
            it.getCompleted()
        }

이후 가져오지 못한 피드의 수를 표시하기위해 xml에 warnings를 추가한다.

    <TextView
        android:id="@+id/warnings"
        android:layout_width="wrap_content"
        android:layout_height="wrap_content"
        android:layout_marginTop="20dp"
        app:layout_constraintTop_toBottomOf="@id/newsCount"
        app:layout_constraintLeft_toLeftOf="parent"
        app:layout_constraintRight_toRightOf="parent"/>

이제 실패한 피드수를 가져온다.

    //각 deferred 내용을 담는 headlines 변수 생성.
    val headlines = requests
        .filter { !it.isCancelled }
        .flatMap {
            it.getCompleted()
        }

    val failed = requests
        .filter { it.isCancelled }
        .size

그리고 UI 를 업데이트 해준다.

    val newsCount = findViewById<TextView>(R.id.newsCount)
    val warnings = findViewById<TextView>(R.id.warnings)
    val obtained = requests.size - failed
    
    launch(Dispatchers.Main) {
        newsCount.text = "Found ${headlines.size} news in $obtained feeds"
        if (failed > 0) {
            warnings.text = "Failed to fetch $failed feeds"
        }
     }

이후 앱을 빌드해보면 아래와 같이 잘 동작하는것을 볼 수 있다.