共享的可变状态与并发

协程可用多线程调度器（比如默认的 Dispatchers.Default）并行执行。这样就可以提出所有常见的并行问题。主要的问题是同步访问共享的可变状态。 协程领域对这个问题的一些解决方案类似于多线程领域中的解决方案， 但其它解决方案则是独一无二的。

问题

我们启动一百个协程，它们都做一千次相同的操作。我们同时会测量它们的完成时间以便进一步的比较：

  1. suspend fun massiveRun(action: suspend () -> Unit) {
  2.     val n = 100  // 启动的协程数量
  3.     val k = 1000 // 每个协程重复执行同一动作的次数
  4.     val time = measureTimeMillis {
  5.         coroutineScope { // 协程的作用域
  6.             repeat(n) {
  7.                 launch {
  8.                     repeat(k) { action() }
  9.                 }
  10.             }
  11.         }
  12.     }
  13.     println("Completed ${n * k} actions in $time ms")    
  14. }

我们从一个非常简单的动作开始：使用多线程的 Dispatchers.Default 来递增一个共享的可变变量。

  1. import kotlinx.coroutines.*
  2. import kotlin.system.*    
  4. suspend fun massiveRun(action: suspend () -> Unit) {
  5.     val n = 100  // 启动的协程数量
  6.     val k = 1000 // 每个协程重复执行同一动作的次数
  7.     val time = measureTimeMillis {
  8.         coroutineScope { // 协程的作用域
  9.             repeat(n) {
  10.                 launch {
  11.                     repeat(k) { action() }
  12.                 }
  13.             }
  14.         }
  15.     }
  16.     println("Completed ${n * k} actions in $time ms")    
  17. }
  19. //sampleStart
  20. var counter = 0
  22. fun main() = runBlocking {
  23.     withContext(Dispatchers.Default) {
  24.         massiveRun {
  25.             counter++
  26.         }
  27.     }
  28.     println("Counter = $counter")
  29. }
  30. //sampleEnd

可以在这里获取完整代码。

共享的可变状态与并发 - 图1

这段代码最后打印出什么结果？它不太可能打印出“Counter = 100000”，因为一百个协程在多个线程中同时递增计数器但没有做并发处理。

volatile 无济于事

有一种常见的误解：volatile 可以解决并发问题。让我们尝试一下：

  1. import kotlinx.coroutines.*
  2. import kotlin.system.*
  4. suspend fun massiveRun(action: suspend () -> Unit) {
  5.     val n = 100  // 启动的协程数量
  6.     val k = 1000 // 每个协程重复执行同一动作的次数
  7.     val time = measureTimeMillis {
  8.         coroutineScope { // 协程的作用域
  9.             repeat(n) {
  10.                 launch {
  11.                     repeat(k) { action() }
  12.                 }
  13.             }
  14.         }
  15.     }
  16.     println("Completed ${n * k} actions in $time ms")    
  17. }
  19. //sampleStart
  20. @Volatile // 在 Kotlin 中 `volatile` 是一个注解
  21. var counter = 0
  23. fun main() = runBlocking {
  24.     withContext(Dispatchers.Default) {
  25.         massiveRun {
  26.             counter++
  27.         }
  28.     }
  29.     println("Counter = $counter")
  30. }
  31. //sampleEnd

可以在这里获取完整代码。

共享的可变状态与并发 - 图2

这段代码运行速度更慢了，但我们仍然不总是最后得到“Counter = 100000”这个结果，因为 volatile 变量保证可线性化（这是“原子”的技术术语）读取和写入变量，但在大量动作（在我们的示例中即“递增”操作）发生时并不提供原子性。

线程安全的数据结构

一种对线程、协程都有效的常规解决方法，就是使用线程安全（也称为同步的、 可线性化、原子）的数据结构，它为需要在共享状态上执行的相应操作提供所有必需的同步处理。在简单的计数器场景中，我们可以使用具有 incrementAndGet 原子操作的 AtomicInteger 类：

  1. import kotlinx.coroutines.*
  2. import java.util.concurrent.atomic.*
  3. import kotlin.system.*
  5. suspend fun massiveRun(action: suspend () -> Unit) {
  6.     val n = 100  // 启动的协程数量
  7.     val k = 1000 // 每个协程重复执行同一动作的次数
  8.     val time = measureTimeMillis {
  9.         coroutineScope { // 协程的作用域
  10.             repeat(n) {
  11.                 launch {
  12.                     repeat(k) { action() }
  13.                 }
  14.             }
  15.         }
  16.     }
  17.     println("Completed ${n * k} actions in $time ms")    
  18. }
  20. //sampleStart
  21. val counter = AtomicInteger()
  23. fun main() = runBlocking {
  24.     withContext(Dispatchers.Default) {
  25.         massiveRun {
  26.             counter.incrementAndGet()
  27.         }
  28.     }
  29.     println("Counter = $counter")
  30. }
  31. //sampleEnd

可以在这里获取完整代码。

共享的可变状态与并发 - 图3

这是针对此类特定问题的最快解决方案。它适用于普通计数器、集合、队列和其他标准数据结构以及它们的基本操作。然而，它并不容易被扩展来应对复杂状态、或一些没有现成的线程安全实现的复杂操作。

以细粒度限制线程

限制线程 是解决共享可变状态问题的一种方案：对特定共享状态的所有访问权都限制在单个线程中。它通常应用于 UI 程序中：所有 UI 状态都局限于单个事件分发线程或应用主线程中。这在协程中很容易实现，通过使用一个单线程上下文。

  1. import kotlinx.coroutines.*
  2. import kotlin.system.*
  4. suspend fun massiveRun(action: suspend () -> Unit) {
  5.     val n = 100  // 启动的协程数量
  6.     val k = 1000 // 每个协程重复执行同一动作的次数
  7.     val time = measureTimeMillis {
  8.         coroutineScope { // 协程的作用域
  9.             repeat(n) {
  10.                 launch {
  11.                     repeat(k) { action() }
  12.                 }
  13.             }
  14.         }
  15.     }
  16.     println("Completed ${n * k} actions in $time ms")    
  17. }
  19. //sampleStart
  20. val counterContext = newSingleThreadContext("CounterContext")
  21. var counter = 0
  23. fun main() = runBlocking {
  24.     withContext(Dispatchers.Default) {
  25.         massiveRun {
  26.             // 将每次自增限制在单线程上下文中
  27.             withContext(counterContext) {
  28.                 counter++
  29.             }
  30.         }
  31.     }
  32.     println("Counter = $counter")
  33. }
  34. //sampleEnd

可以在这里获取完整代码。

共享的可变状态与并发 - 图4

这段代码运行非常缓慢，因为它进行了 细粒度 的线程限制。每个增量操作都得使用 [withContext(counterContext)] 块从多线程 Dispatchers.Default 上下文切换到单线程上下文。

以粗粒度限制线程

在实践中，线程限制是在大段代码中执行的，例如：状态更新类业务逻辑中大部分都是限于单线程中。下面的示例演示了这种情况， 在单线程上下文中运行每个协程。

  1. import kotlinx.coroutines.*
  2. import kotlin.system.*
  4. suspend fun massiveRun(action: suspend () -> Unit) {
  5.     val n = 100  // 启动的协程数量
  6.     val k = 1000 // 每个协程重复执行同一动作的次数
  7.     val time = measureTimeMillis {
  8.         coroutineScope { // 协程的作用域
  9.             repeat(n) {
  10.                 launch {
  11.                     repeat(k) { action() }
  12.                 }
  13.             }
  14.         }
  15.     }
  16.     println("Completed ${n * k} actions in $time ms")    
  17. }
  19. //sampleStart
  20. val counterContext = newSingleThreadContext("CounterContext")
  21. var counter = 0
  23. fun main() = runBlocking {
  24.     // 将一切都限制在单线程上下文中
  25.     withContext(counterContext) {
  26.         massiveRun {
  27.             counter++
  28.         }
  29.     }
  30.     println("Counter = $counter")
  31. }
  32. //sampleEnd

可以在这里获取完整代码。

共享的可变状态与并发 - 图5

这段代码运行更快而且打印出了正确的结果。

互斥

该问题的互斥解决方案：使用永远不会同时执行的 关键代码块 来保护共享状态的所有修改。在阻塞的世界中，你通常会为此目的使用 synchronized 或者 ReentrantLock。 在协程中的替代品叫做 Mutex 。它具有 lock 和 unlock 方法， 可以隔离关键的部分。关键的区别在于 Mutex.lock() 是一个挂起函数，它不会阻塞线程。

还有 withLock 扩展函数，可以方便的替代常用的 mutex.lock(); try { …… } finally { mutex.unlock() } 模式：

  1. import kotlinx.coroutines.*
  2. import kotlinx.coroutines.sync.*
  3. import kotlin.system.*
  5. suspend fun massiveRun(action: suspend () -> Unit) {
  6.     val n = 100  // 启动的协程数量
  7.     val k = 1000 // 每个协程重复执行同一动作的次数
  8.     val time = measureTimeMillis {
  9.         coroutineScope { // 协程的作用域
  10.             repeat(n) {
  11.                 launch {
  12.                     repeat(k) { action() }
  13.                 }
  14.             }
  15.         }
  16.     }
  17.     println("Completed ${n * k} actions in $time ms")    
  18. }
  20. //sampleStart
  21. val mutex = Mutex()
  22. var counter = 0
  24. fun main() = runBlocking {
  25.     withContext(Dispatchers.Default) {
  26.         massiveRun {
  27.             // 用锁保护每次自增
  28.             mutex.withLock {
  29.                 counter++
  30.             }
  31.         }
  32.     }
  33.     println("Counter = $counter")
  34. }
  35. //sampleEnd

可以在这里获取完整代码。

共享的可变状态与并发 - 图6

此示例中锁是细粒度的，因此会付出一些代价。但是对于某些必须定期修改共享状态的场景，它是一个不错的选择，但是没有自然线程可以限制此状态。