[TOC] ## 共享的可變狀態與并發 協程可用多線程調度器（比如默認的 [Dispatchers.Default](https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-dispatchers/-default.html)）并發執行。這樣就可以提出所有常見的并發問題。主要的問題是同步訪問**共享的可變狀態**。協程領域對這個問題的一些解決方案類似于多線程領域中的解決方案，但其它解決方案則是獨一無二的。 ### 問題 我們啟動一百個協程，它們都做一千次相同的操作。我們同時會測量它們的完成時間以便進一步的比較： ```kotlin suspend fun massiveRun(action: suspend () -> Unit) { val n = 100 // 啟動的協程數量 val k = 1000 // 每個協程重復執行同一動作的次數 val time = measureTimeMillis { coroutineScope { // 協程的作用域 repeat(n) { launch { repeat(k) { action() } } } } } println("Completed ${n * k} actions in $time ms") } ``` 我們從一個非常簡單的動作開始：使用多線程的 [Dispatchers.Default](https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-dispatchers/-default.html) 來遞增一個共享的可變變量。 ```kotlin import kotlinx.coroutines.* import kotlin.system.* suspend fun massiveRun(action: suspend () -> Unit) { val n = 100 // 啟動的協程數量 val k = 1000 // 每個協程重復執行同一動作的次數 val time = measureTimeMillis { coroutineScope { // 協程的作用域 repeat(n) { launch { repeat(k) { action() } } } } } println("Completed ${n * k} actions in $time ms") } //sampleStart var counter = 0 fun main() = runBlocking { withContext(Dispatchers.Default) { massiveRun { counter++ } } println("Counter = $counter") } //sampleEnd ``` > 可以在[這里](https://github.com/hltj/kotlinx.coroutines-cn/blob/master/kotlinx-coroutines-core/jvm/test/guide/example-sync-01.kt)獲取完整代碼。 這段代碼最后打印出什么結果？它不太可能打印出“Counter = 100000”，因為一百個協程在多個線程中同時遞增計數器但沒有做并發處理。 ### volatile 無濟于事 有一種常見的誤解：volatile 可以解決并發問題。讓我們嘗試一下： ```kotlin import kotlinx.coroutines.* import kotlin.system.* suspend fun massiveRun(action: suspend () -> Unit) { val n = 100 // 啟動的協程數量 val k = 1000 // 每個協程重復執行同一動作的次數 val time = measureTimeMillis { coroutineScope { // 協程的作用域 repeat(n) { launch { repeat(k) { action() } } } } } println("Completed ${n * k} actions in $time ms") } //sampleStart @Volatile // 在 Kotlin 中 `volatile` 是一個注解 var counter = 0 fun main() = runBlocking { withContext(Dispatchers.Default) { massiveRun { counter++ } } println("Counter = $counter") } //sampleEnd ``` > 可以在[這里](https://github.com/hltj/kotlinx.coroutines-cn/blob/master/kotlinx-coroutines-core/jvm/test/guide/example-sync-02.kt)獲取完整代碼。 這段代碼運行速度更慢了，但我們最后仍然沒有得到“Counter = 100000”這個結果，因為 volatile 變量保證可線性化（這是“原子”的技術術語）讀取和寫入變量，但在大量動作（在我們的示例中即“遞增”操作）發生時并不提供原子性。 ### 線程安全的數據結構 一種對線程、協程都有效的常規解決方法，就是使用線程安全（也稱為同步的、可線性化、原子）的數據結構，它為需要在共享狀態上執行的相應操作提供所有必需的同步處理。在簡單的計數器場景中，我們可以使用具有`incrementAndGet` 原子操作的 `AtomicInteger` 類： ```kotlin import kotlinx.coroutines.* import java.util.concurrent.atomic.* import kotlin.system.* suspend fun massiveRun(action: suspend () -> Unit) { val n = 100 // 啟動的協程數量 val k = 1000 // 每個協程重復執行同一動作的次數 val time = measureTimeMillis { coroutineScope { // 協程的作用域 repeat(n) { launch { repeat(k) { action() } } } } } println("Completed ${n * k} actions in $time ms") } //sampleStart var counter = AtomicInteger() fun main() = runBlocking { withContext(Dispatchers.Default) { massiveRun { counter.incrementAndGet() } } println("Counter = $counter") } //sampleEnd ``` > 可以在[這里](https://github.com/hltj/kotlinx.coroutines-cn/blob/master/kotlinx-coroutines-core/jvm/test/guide/example-sync-03.kt)獲取完整代碼。 這是針對此類特定問題的最快解決方案。它適用于普通計數器、集合、隊列和其他標準數據結構以及它們的基本操作。然而，它并不容易被擴展來應對復雜狀態、或一些沒有現成的線程安全實現的復雜操作。 ### 以細粒度限制線程 _限制線程_ 是解決共享可變狀態問題的一種方案：對特定共享狀態的所有訪問權都限制在單個線程中。它通常應用于 UI 程序中：所有 UI 狀態都局限于單個事件分發線程或應用主線程中。這在協程中很容易實現，通過使用一個單線程上下文： ```kotlin import kotlinx.coroutines.* import kotlin.system.* suspend fun massiveRun(action: suspend () -> Unit) { val n = 100 // 啟動的協程數量 val k = 1000 // 每個協程重復執行同一動作的次數 val time = measureTimeMillis { coroutineScope { // 協程的作用域 repeat(n) { launch { repeat(k) { action() } } } } } println("Completed ${n * k} actions in $time ms") } //sampleStart val counterContext = newSingleThreadContext("CounterContext") var counter = 0 fun main() = runBlocking { withContext(Dispatchers.Default) { massiveRun { // 將每次自增限制在單線程上下文中 withContext(counterContext) { counter++ } } } println("Counter = $counter") } //sampleEnd ``` > 可以在[這里](https://github.com/hltj/kotlinx.coroutines-cn/blob/master/kotlinx-coroutines-core/jvm/test/guide/example-sync-04.kt)獲取完整代碼。 這段代碼運行非常緩慢，因為它進行了 _細粒度_ 的線程限制。每個增量操作都得使用[withContext(counterContext)] 塊從多線程 [Dispatchers.Default](https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-dispatchers/-default.html) 上下文切換到單線程上下文。 ### 以粗粒度限制線程 在實踐中，線程限制是在大段代碼中執行的，例如：狀態更新類業務邏輯中大部分都是限于單線程中。下面的示例演示了這種情況，在單線程上下文中運行每個協程。 ```kotlin import kotlinx.coroutines.* import kotlin.system.* suspend fun massiveRun(action: suspend () -> Unit) { val n = 100 // 啟動的協程數量 val k = 1000 // 每個協程重復執行同一動作的次數 val time = measureTimeMillis { coroutineScope { // 協程的作用域 repeat(n) { launch { repeat(k) { action() } } } } } println("Completed ${n * k} actions in $time ms") } //sampleStart val counterContext = newSingleThreadContext("CounterContext") var counter = 0 fun main() = runBlocking { // 將一切都限制在單線程上下文中 withContext(counterContext) { massiveRun { counter++ } } println("Counter = $counter") } //sampleEnd ``` > 可以在[這里](https://github.com/hltj/kotlinx.coroutines-cn/blob/master/kotlinx-coroutines-core/jvm/test/guide/example-sync-05.kt)獲取完整代碼。 ``` Completed 100000 actions in xxx ms Counter = 100000 ``` 這段代碼運行更快而且打印出了正確的結果。 ### 互斥 該問題的互斥解決方案：使用永遠不會同時執行的 _關鍵代碼塊_來保護共享狀態的所有修改。在阻塞的世界中，你通常會為此目的使用 `synchronized` 或者 `ReentrantLock`。在協程中的替代品叫做 [Mutex](https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.sync/-mutex/index.html) 。它具有 [lock](https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.sync/-mutex/lock.html)和 [unlock](https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.sync/-mutex/unlock.html)方法，可以隔離關鍵的部分。關鍵的區別在于 `Mutex.lock()` 是一個掛起函數，它不會阻塞線程。 還有 [withLock](https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.sync/with-lock.html) 擴展函數，可以方便的替代常用的`mutex.lock(); try { …… } finally { mutex.unlock() }` 模式： ```kotlin import kotlinx.coroutines.* import kotlinx.coroutines.sync.* import kotlin.system.* suspend fun massiveRun(action: suspend () -> Unit) { val n = 100 // 啟動的協程數量 val k = 1000 // 每個協程重復執行同一動作的次數 val time = measureTimeMillis { coroutineScope { // 協程的作用域 repeat(n) { launch { repeat(k) { action() } } } } } println("Completed ${n * k} actions in $time ms") } //sampleStart val mutex = Mutex() var counter = 0 fun main() = runBlocking { withContext(Dispatchers.Default) { massiveRun { // 用鎖保護每次自增 mutex.withLock { counter++ } } } println("Counter = $counter") } //sampleEnd ``` > 可以在[這里](https://github.com/hltj/kotlinx.coroutines-cn/blob/master/kotlinx-coroutines-core/jvm/test/guide/example-sync-06.kt)獲取完整代碼。 ``` Completed 100000 actions in xxx ms Counter = 100000 ``` 此示例中鎖是細粒度的，因此會付出一些代價。但是對于某些必須定期修改共享狀態的場景，它是一個不錯的選擇，但是沒有自然線程可以限制此狀態。 ### Actors 一個 [actor](https://en.wikipedia.org/wiki/Actor_model) 是由協程、被限制并封裝到該協程中的狀態以及一個與其它協程通信的 _通道_ 組合而成的一個實體。一個簡單的 actor 可以簡單的寫成一個函數，但是一個擁有復雜狀態的 actor 更適合由類來表示。 有一個 [actor](https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/actor.html) 協程構建器，它可以方便地將 actor 的郵箱通道組合到其作用域中（用來接收消息）、組合發送 channel 與結果集對象，這樣對 actor 的單個引用就可以作為其句柄持有。 使用 actor 的第一步是定義一個 actor 要處理的消息類。Kotlin 的[密封類](https://kotlinlang.org/docs/reference/sealed-classes.html)很適合這種場景。我們使用 `IncCounter` 消息（用來遞增計數器）和 `GetCounter` 消息（用來獲取值）來定義 `CounterMsg` 密封類。后者需要發送回復。[CompletableDeferred](https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-completable-deferred/index.html) 通信原語表示未來可知（可傳達）的單個值，這里被用于此目的。 ```kotlin // 計數器 Actor 的各種類型 sealed class CounterMsg object IncCounter : CounterMsg() // 遞增計數器的單向消息 class GetCounter(val response: CompletableDeferred<Int>) : CounterMsg() // 攜帶回復的請求 ``` 接下來我們定義一個函數，使用 [actor](https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/actor.html) 協程構建器來啟動一個 actor： ```kotlin // 這個函數啟動一個新的計數器 actor fun CoroutineScope.counterActor() = actor<CounterMsg> { var counter = 0 // actor 狀態 for (msg in channel) { // 即將到來消息的迭代器 when (msg) { is IncCounter -> counter++ is GetCounter -> msg.response.complete(counter) } } } ``` main 函數代碼很簡單： ```kotlin import kotlinx.coroutines.* import kotlinx.coroutines.channels.* import kotlin.system.* suspend fun massiveRun(action: suspend () -> Unit) { val n = 100 // 啟動的協程數量 val k = 1000 // 每個協程重復執行同個動作的次數 val time = measureTimeMillis { coroutineScope { // 協程的作用域 repeat(n) { launch { repeat(k) { action() } } } } } println("Completed ${n * k} actions in $time ms") } // 計數器 Actor 的各種類型 sealed class CounterMsg object IncCounter : CounterMsg() // 遞增計數器的單向消息 class GetCounter(val response: CompletableDeferred<Int>) : CounterMsg() // 攜帶回復的請求 // 這個函數啟動一個新的計數器 actor fun CoroutineScope.counterActor() = actor<CounterMsg> { var counter = 0 // actor 狀態 for (msg in channel) { // 即將到來消息的迭代器 when (msg) { is IncCounter -> counter++ is GetCounter -> msg.response.complete(counter) } } } //sampleStart fun main() = runBlocking<Unit> { val counter = counterActor() // 創建該 actor withContext(Dispatchers.Default) { massiveRun { counter.send(IncCounter) } } // 發送一條消息以用來從一個 actor 中獲取計數值 val response = CompletableDeferred<Int>() counter.send(GetCounter(response)) println("Counter = ${response.await()}") counter.close() // 關閉該actor } //sampleEnd ``` > 可以在[這里](https://github.com/hltj/kotlinx.coroutines-cn/blob/master/kotlinx-coroutines-core/jvm/test/guide/example-sync-07.kt)獲取完整代碼。 ``` Completed 100000 actions in xxx ms Counter = 100000 ``` actor 本身執行時所處上下文（就正確性而言）無關緊要。一個 actor 是一個協程，而一個協程是按順序執行的，因此將狀態限制到特定協程可以解決共享可變狀態的問題。實際上，actor 可以修改自己的私有狀態，但只能通過消息互相影響（避免任何鎖定）。 actor 在高負載下比鎖更有效，因為在這種情況下它總是有工作要做，而且根本不需要切換到不同的上下文。 > 注意，[actor](https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/actor.html) 協程構建器是一個雙重的 [produce](https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/produce.html) 協程構建器。一個 actor 與它接收消息的通道相關聯，而一個 producer 與它發送元素的通道相關聯。 [Dispatchers.Default]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-dispatchers/-default.html [withContext]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/with-context.html [CompletableDeferred]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-completable-deferred/index.html [Mutex]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.sync/-mutex/index.html [Mutex.lock]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.sync/-mutex/lock.html [Mutex.unlock]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.sync/-mutex/unlock.html [withLock]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.sync/with-lock.html [actor]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/actor.html [produce]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/produce.html