Day8：結構化併發 (Structured Concurrency)

13th鐵人賽 kotlin kotlin讀書會 strctured concurrency

Andy 安迪

團隊Kotlin 愛台灣 2021

2021-09-13 22:48:29

1852 瀏覽

分享至

還記得我們第一個 Coroutine 程式嗎？

suspend fun showContents() = coroutineScope {
    launch {
        val token = login(userName, password)
        val content = fetch(token)
        withContext(Dispatchers.Main) {
            showContent(content)
        }
    }
}

在 launch 裏面的 suspend 函式 login() 以及 fetch() 會繼承 launch 的 coroutine context，所以會這兩個 suspend 函式都會用相同的 coroutine context 來執行，會使用相同的調度器(Dispatchers)，由於我們沒有指定，所以這邊我們會使用 Dispatchers.Default ，也就是背景執行。

Kotlin 的 Coroutine 是採用結構化的併發，在這段程式碼的呼叫，會有一段起始點，在最外層的我稱為父 Coroutine ，在內層的我稱為子 Coroutine。父 Coroutine 會等到所有的子 Coroutine 完成之後才會結束。

所以上面的執行時間，會從第一個 login() 的執行時間到 showContent() 的執行時間。

在上一篇文章中，我們知道我們可以呼叫 Job 的 cancel() 來取消當下的任務，但是如果同時執行多個任務，也就是併發任務，那我們針對某個 Job 呼叫 cancel() 會發生什麼事呢？

取消 Job

我們知道 Kotlin 的 suspend 函式必須要在 Coroutine Scope 內才能執行，有的時候我們可能需要在一個範圍內執行多個 Coroutine，如下：

fun main() = runBlocking {
    val job1 = launch {
        delay(100)
        println("Job1 done")
    }

    val job2 = launch {
        delay(1000)
        println("Job2 done")
    }
    println("Outer coroutine done")
}

Concurrency

第一個 launch 耗時 100 毫秒，第二個 launch 耗時 1000 毫秒，如同我們之前提到的，因為這兩個 launch 是在 runBlocking 所產生的 Coroutine Scope 內，所以會先執行外層的程式，並且外層會等到子 Coroutine 都完成任務時才結束。

所以上方範例的結果會先顯示出 Outer coroutine done ，接著才是 Job1 done以及 Job2 done。

假如 job2 所花費的時間已經超出預期，我們可以主動呼叫 job2.cancel() 來把 job2 給停掉。

fun main() = runBlocking {
    val job1 = launch {
        delay(100)
        println("Job1 done")
    }

    val job2 = launch() {
        delay(1000)
        println("Job2 done")
    }
    println("Outer coroutine done")
    delay(300)
    job2.cancel()
}

cancel

→ 只顯示了 Outer coroutine done 以及 Job1 done ，job2 則是被取消。

假設所有的子 job 需要被取消呢？我們該怎麼處理？

方法1：呼叫所有 job 的 `cancel()`

fun main() = runBlocking {
    val job1 = launch {
        delay(1000)
        println("Job1 done")
    }

    val job2 = launch() {
        delay(1000)
        println("Job2 done")
    }
    println("Outer coroutine done")
    delay(300)
    job1.cancel()
    job2.cancel()
}

Cancel all

→ 如同我們前面介紹的，我們可以呼叫 job 的 cancel()來取消 job，所以需要取消所有的 job，就呼叫所有的 cancel() 。

方法2：取消父 coroutine

用一個 launch 包住這兩個 launch讓這原本的 coroutine 成為這個 launch 的子 coroutine，調用父 coroutine 的 cancel()

fun main() = runBlocking {
    val job = launch {
        val job1 = launch {
            delay(1000)
            println("Job1 done")
        }
        val job2 = launch {
            delay(1000)
            println("Job2 done")
        }
    }
    println("Outer coroutine done")
    delay(300)
    job.cancel()
}

→因為 Kotlin 的 coroutine 是有階層的，當父 coroutine 被取消後，子 coroutine 也會同時被取消。

Nested coroutine

runBlocking 所產生的 Coroutine 為 BlockingCoroutine。

→ 所以我們只要呼叫 CoroutineScope 的 cancel() 就能夠取消所有在這個 CoroutineScope 的 Job了。

小結

在結構化併發的架構下，父 Coroutine 會等到全部的子 Coroutine 都結束之後才會結束。而在這樣的架構之下，如果沒有特別設定子 Coroutine 的 coroutine context，就會繼承父 Coroutine 的 context。父 Coroutine 被取消之後，所有的子 Coroutine 也會一併被取消，這樣子的設計就不會發生當較高的階層被取消後，較低的階層還在運行，然後發生錯誤。

同樣的，如果同一層的 Coroutine 有一個 Job 被取消，在後面尚未執行完成的 Job 也會同時被取消。