22、协程与多线程并发：Mutex互斥锁、Semaphore信号量、协程安全的共享状态、Actor模式在协程中的实现

并发编程，说白了就是多个任务同时访问同一块数据。在Android开发中，这几乎是绕不开的坎。Kotlin协程虽然简化了异步代码，但并发安全问题依然存在。今天我们就来聊聊协程世界里，怎么优雅地解决这些问题。

一、协程并发问题的本质

先问一个问题：协程是运行在单线程里的吗？当然不是。协程可以切换线程，也可以多个协程同时运行。当多个协程同时读写同一个变量时，就会产生竞态条件。

我在项目中遇到过这样一个场景：一个计数器，多个协程同时递增，结果总是少几个数。这就是典型的并发问题。协程本身不解决数据竞争，它只提供了一套挂起机制。真正的并发控制，需要我们自己来。

二、Mutex互斥锁：协程版的synchronized

Mutex是协程中的互斥锁。它和传统的Lock不同，Mutex的lock()是一个挂起函数，不会阻塞线程。这意味着，当一个协程等待锁时，线程可以去执行其他协程。

// 基本用法
val mutex = Mutex()
var sharedCounter = 0

suspend fun increment() {
    mutex.withLock {
        sharedCounter++
    }
}

// 多个协程并发调用
coroutineScope {
    repeat(1000) {
        launch {
            increment()
        }
    }
}
// 最终sharedCounter == 1000，安全

我个人习惯用withLock而不是手动lock/unlock。为什么？因为withLock保证了无论代码块是否抛出异常，锁都会被释放。手动管理容易漏掉finally块。

三、Semaphore信号量：控制并发数量

Mutex只允许一个协程访问资源。但有时候，我们希望限制并发数量，比如同时最多3个网络请求。这时候Semaphore就派上用场了。

// 创建一个允许3个并发访问的信号量
val semaphore = Semaphore(3)

suspend fun limitedAccess(taskId: Int) {
    semaphore.withPermit {
        println("Task $taskId 开始执行")
        delay(1000) // 模拟耗时操作
        println("Task $taskId 执行完毕")
    }
}

// 启动10个协程
coroutineScope {
    repeat(10) {
        launch {
            limitedAccess(it)
        }
    }
}
// 同一时刻最多只有3个任务在执行

我曾经用Semaphore做过一个限流器。后端接口只能承受每秒5个请求，我用Semaphore配合协程，完美控制了并发数。注意，Semaphore的acquire()也是挂起函数，不会阻塞线程。

四、协程安全的共享状态

除了Mutex和Semaphore，Kotlin协程还提供了几种更高级的共享状态方案。

4.1 使用原子类

对于简单的计数器或标志位，原子类是最轻量的选择。

val atomicCounter = AtomicInteger(0)

suspend fun atomicIncrement() {
    atomicCounter.incrementAndGet()
}
// 不需要锁，性能更好

但原子类只适用于单一变量的操作。如果你需要同时更新多个变量，原子类就不够用了。

4.2 使用ConcurrentHashMap

对于集合类的共享状态，ConcurrentHashMap是首选。它的读写操作都是线程安全的。

val cache = ConcurrentHashMap<String, Any>()

suspend fun getOrPut(key: String, value: () -> Any): Any {
    return cache.getOrPut(key) { value() }
}

4.3 使用协程局部变量

有时候，状态不需要跨协程共享，只需要在协程内部传递。这时候可以用CoroutineContext来保存状态。

class UserContext(val userId: String) : CoroutineContext.Element {
    companion object Key : CoroutineContext.Key<UserContext>
    override val key: CoroutineContext.Key<*> get() = Key
}

// 在协程中传递
launch(UserContext("123")) {
    val ctx = coroutineContext[UserContext.Key]
    println("当前用户: ${ctx?.userId}")
}

嗯，这里要注意：协程局部变量只对当前协程及其子协程可见，不能跨协程共享。适合用来传递请求ID、用户信息等上下文数据。

五、Actor模式在协程中的实现

Actor模式是一种经典的并发模型。它的核心思想是：每个Actor拥有自己的状态，外部只能通过发送消息来与Actor交互。Actor内部保证消息的串行处理，天然避免了并发问题。

Kotlin协程没有内置的Actor，但我们可以用Channel轻松实现。

// 定义一个Actor
class CounterActor {
    private val scope = CoroutineScope(Dispatchers.Default + SupervisorJob())
    private val channel = Channel<CounterMsg>(Channel.UNLIMITED)
    private var count = 0

    sealed class CounterMsg {
        data class Increment(val amount: Int) : CounterMsg()
        data class GetCount(val response: CompletableDeferred<Int>) : CounterMsg()
    }

    init {
        scope.launch {
            for (msg in channel) {
                when (msg) {
                    is CounterMsg.Increment -> count += msg.amount
                    is CounterMsg.GetCount -> msg.response.complete(count)
                }
            }
        }
    }

    suspend fun increment(amount: Int) {
        channel.send(CounterMsg.Increment(amount))
    }

    suspend fun getCount(): Int {
        val response = CompletableDeferred<Int>()
        channel.send(CounterMsg.GetCount(response))
        return response.await()
    }

    fun close() {
        channel.close()
        scope.cancel()
    }
}

// 使用Actor
suspend fun main() {
    val actor = CounterActor()
    coroutineScope {
        repeat(1000) {
            launch {
                actor.increment(1)
            }
        }
    }
    println("最终计数: ${actor.getCount()}") // 1000
    actor.close()
}

我在项目中用Actor模式管理WebSocket连接状态。每个连接对应一个Actor，消息按顺序处理，状态永远不会乱。你想想看，如果不用Actor，多个协程同时修改连接状态，那调试起来得多痛苦。

六、如何选择？一张图看懂

说了这么多，到底该用哪个？我画了一张图，帮你快速决策。

七、实战建议与避坑

最后，分享几个我在实战中总结的经验。

场景	推荐方案	不推荐
简单计数器	AtomicInteger	Mutex（太重）
缓存读写	ConcurrentHashMap	全局锁（性能差）
资源池限流	Semaphore	自己写计数器（容易出错）
复杂业务状态	Mutex 或 Actor	synchronized（阻塞线程）
跨协程消息传递	Channel + Actor	共享变量（竞态条件）

另外，Actor模式虽然优雅，但也不是银弹。如果你的状态逻辑很简单，用Mutex就够了。Actor的Channel会有额外的内存开销。我一般只在状态复杂、需要消息序列化的场景下才用Actor。

好了，关于协程并发控制，今天就聊到这里。记住一个原则：能用原子类就别用锁，能用Mutex就别用Actor。选择最轻量的方案，代码才够优雅。

---

公众号:蓝海资料掘金营，微信deep3321