Kotlin互斥锁(Mutex)：协程的线程安全守护神

本文译自「Kotlin Mutex: Thread-Safe Concurrency for Coroutines」，原文链接https://carrion.dev/en/posts/kotlin-mutex-concurrency-guide/，由Ignacio Carrión发布于2025年10月3日。

使用 Kotlin 协程构建并发应用程序时，保护共享的可变状态至关重要。虽然传统的 Java 同步工具（例如 synchronized 块和 ReentrantLock）可以正常工作，但它们会阻塞线程，并且与协程的挂起模型不兼容。因此，引入 Mutex——一个协程友好的同步原语，它提供互斥而不阻塞线程。

本指南探讨了何时使用 Mutex、最佳实践以及它与其他并发控制机制的比较。

TL;DR：省流版本的建议

当需要保护多个协程访问的共享可变状态时，请使用 Mutex。
在协程代码中，优先使用 Mutex 而不是 synchronized，以避免阻塞线程。
使用 mutex.withLock { } 自动获取和释放锁。
对于更复杂的状态管理场景，请考虑使用 Actor 或 StateFlow。
对于简单的计数器，请改用 AtomicInteger 或 AtomicReference。
如果需要将并发访问限制为多个许可，请使用 Semaphore。
如果不使用 withLock，请始终在 finally 块中释放锁。

什么是互斥锁？

Mutex（互斥）是 kotlinx.coroutines 中的同步原语，用于确保同一时间只有一个协程可以执行临界区。与阻塞线程的传统锁不同，Mutex 会暂停协程，从而使线程可以自由地执行其他工作。

基本结构：

import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock

val mutex = Mutex()

suspend fun protectedOperation() {
    mutex.withLock {
        // Critical section - only one coroutine at a time
        // Modify shared state safely here
    }
}

关键特性：

非阻塞：暂停协程而不是阻塞线程
公平：默认按先进先出顺序授予访问权限
不重入安全：持有锁的协程无法再次获取锁（防止死锁）
轻量级：比线程阻塞锁更高效

互斥锁的核心用例

最常见的用例——确保对共享变量的安全访问：

class CounterService {
    private var counter = 0
    private val mutex = Mutex()

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

    suspend fun getCount(): Int {
        return mutex.withLock {
            counter
        }
    }
}

2. 协调资源访问

当多个协程需要对某个资源进行独占访问时：

class FileWriter(private val file: File) {
    private val mutex = Mutex()

    suspend fun appendLine(line: String) {
        mutex.withLock {
            file.appendText("$line\n")
        }
    }
}

3.确保顺序执行

即使操作是并发触发的，也必须按顺序执行：

class OrderProcessor {
    private val mutex = Mutex()
    private val orders = mutableListOf<Order>()

    suspend fun processOrder(order: Order) {
        mutex.withLock {
            // Ensure orders are processed sequentially
            orders.add(order)
            validateOrder(order)
            persistOrder(order)
        }
    }
}

4. 线程安全的延迟初始化

在挂起上下文中实现线程安全的延迟初始化：

class DatabaseConnection {
    private var connection: Connection? = null
    private val mutex = Mutex()

    suspend fun getConnection(): Connection {
        if (connection != null) return connection!!

        return mutex.withLock {
            // Double-check inside lock
            connection ?: createConnection().also { connection = it }
        }
    }

    private suspend fun createConnection(): Connection {
        delay(1000) // Simulate connection setup
        return Connection()
    }
}

最佳实践

1. 始终使用 withLock

即使发生异常，withLock 也会自动处理锁的获取和释放：

// ✅ Good: Automatic cleanup
mutex.withLock {
    dangerousOperation()
}

// ❌ Bad: Manual management, error-prone
mutex.lock()
try {
    dangerousOperation()
} finally {
    mutex.unlock()
}

2. 保持临界区较小

尽量减少锁的持有时间以减少争用：

// ✅ Good: Lock only for critical section
suspend fun updateUser(userId: String, name: String) {
    val validated = validateName(name) // Outside lock

    mutex.withLock {
        userCache[userId] = validated // Only this needs protection
    }

    notifyObservers(userId) // Outside lock
}

// ❌ Bad: Holding lock during slow operations
suspend fun updateUserSlow(userId: String, name: String) {
    mutex.withLock {
        val validated = validateName(name) // Slow operation inside lock
        userCache[userId] = validated
        notifyObservers(userId) // I/O inside lock
    }
}

3. 避免嵌套锁

互斥锁不可重入。避免两次获取同一个锁：

// ❌ Bad: Deadlock!
suspend fun problematic() {
    mutex.withLock {
        helperFunction() // Tries to acquire mutex again
    }
}

suspend fun helperFunction() {
    mutex.withLock {
        // Will suspend forever
    }
}

// ✅ Good: Restructure to avoid nesting
suspend fun better() {
    mutex.withLock {
        helperFunctionUnsafe() // No lock acquisition
    }
}

fun helperFunctionUnsafe() {
    // Assumes caller holds lock
}

4. 优先考虑无锁替代方案

对于简单操作，原子类型速度更快：

// ✅ Better for simple counters
class AtomicCounter {
    private val counter = AtomicInteger(0)

    fun increment() = counter.incrementAndGet()
    fun get() = counter.get()
}

// ❌ Overkill for a simple counter
class MutexCounter {
    private var counter = 0
    private val mutex = Mutex()

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

5.文档锁不变量

明确锁保护的对象：

class UserCache {
    private val mutex = Mutex() // Protects userMap and lastUpdate
    private val userMap = mutableMapOf<String, User>()
    private var lastUpdate = 0L

    suspend fun updateUser(id: String, user: User) {
        mutex.withLock {
            userMap[id] = user
            lastUpdate = System.currentTimeMillis()
        }
    }
}

互斥锁 vs. 其他同步方法

互斥锁 vs. synchronized

// Traditional synchronized (blocks thread)
class SynchronizedCounter {
    private var count = 0

    @Synchronized
    fun increment() {
        count++ // Thread blocked while waiting
    }
}

// Mutex (suspends coroutine)
class MutexCounter {
    private var count = 0
    private val mutex = Mutex()

    suspend fun increment() {
        mutex.withLock {
            count++ // Coroutine suspended, thread free
        }
    }
}

何时该用哪个：

对于非暂停代码和旧版 Java 互操作，请使用 synchronized
对于暂停函数和基于协程的代码，请使用 Mutex
在协程上下文中，Mutex 效率更高，因为线程不会被阻塞

互斥锁 vs. 信号量

// Mutex: Only one coroutine at a time
val mutex = Mutex()

// Semaphore: N coroutines at a time
val semaphore = Semaphore(permits = 3)

// Example: Rate limiting API calls
class ApiClient {
    private val semaphore = Semaphore(5) // Max 5 concurrent requests

    suspend fun makeRequest(endpoint: String): Response {
        semaphore.withPermit {
            return httpClient.get(endpoint)
        }
    }
}

何时使用谁：

需要独占访问（单次许可）时使用 Mutex
需要将并发限制为 N 个操作时使用 Semaphore

互斥锁 vs. Actor

// Mutex: Manual synchronization
class MutexBasedCache {
    private val cache = mutableMapOf<String, Data>()
    private val mutex = Mutex()

    suspend fun get(key: String) = mutex.withLock { cache[key] }
    suspend fun put(key: String, value: Data) = mutex.withLock { cache[key] = value }
}

// Actor: Message-based synchronization
sealed class CacheMessage
data class Get(val key: String, val response: CompletableDeferred<Data?>) : CacheMessage()
data class Put(val key: String, val value: Data) : CacheMessage()

fun CoroutineScope.cacheActor() = actor<CacheMessage> {
    val cache = mutableMapOf<String, Data>()

    for (msg in channel) {
        when (msg) {
            is Get -> msg.response.complete(cache[msg.key])
            is Put -> cache[msg.key] = msg.value
        }
    }
}

何时使用谁：

使用 Mutex 进行直接方法调用的简单同步
对于复杂的状态机或需要消息队列时，使用 Actor
Actor 提供更好的封装性，并且可以处理背压

Mutex 与 StateFlow

// Mutex: Imperative state management
class MutexState {
    private var state = 0
    private val mutex = Mutex()

    suspend fun updateState(transform: (Int) -> Int) {
        mutex.withLock {
            state = transform(state)
        }
    }
}

// StateFlow: Reactive state management
class FlowState {
    private val _state = MutableStateFlow(0)
    val state: StateFlow<Int> = _state.asStateFlow()

    fun updateState(transform: (Int) -> Int) {
        _state.update(transform) // Thread-safe built-in
    }
}

何时使用哪个：

需要自定义同步逻辑时使用 Mutex
使用 StateFlow 进行内置线程安全的可观察状态
StateFlow 更适合 UI 状态和响应式架构

Mutex 与原子类型

// AtomicInteger: Lock-free for simple operations
class AtomicCounter {
    private val counter = AtomicInteger(0)

    fun increment() = counter.incrementAndGet()
    fun addAndGet(delta: Int) = counter.addAndGet(delta)
}

// Mutex: For complex operations
class ComplexCounter {
    private var counter = 0
    private var history = mutableListOf<Int>()
    private val mutex = Mutex()

    suspend fun increment() {
        mutex.withLock {
            counter++
            history.add(counter) // Multiple operations
        }
    }
}

何时使用哪个：

使用原子类型进行单变量操作（计数器、标志）
需要协调多个变量时使用 Mutex
原子操作速度更快，但受限于特定操作

常见陷阱

1. 忘记使用 suspend

互斥操作需要暂停：

// ❌ Won't compile
fun broken() {
    mutex.withLock { } // Error: suspend function called in non-suspend context
}

// ✅ Correct
suspend fun correct() {
    mutex.withLock { }
}

2. 长时间操作期间持有锁

// ❌ Bad: Holding lock during I/O
suspend fun bad(url: String) {
    mutex.withLock {
        val data = httpClient.get(url) // Network call inside lock
        cache[url] = data
    }
}

// ✅ Good: Fetch outside lock
suspend fun good(url: String) {
    val data = httpClient.get(url)
    mutex.withLock {
        cache[url] = data
    }
}

3. 假设可重入

// ❌ Deadlock: Mutex is not reentrant
suspend fun outer() {
    mutex.withLock {
        inner() // Deadlock!
    }
}

suspend fun inner() {
    mutex.withLock {
        // Never reached
    }
}

4. 不处理取消

持有锁时务必考虑取消：

// ✅ Good: withLock handles cancellation
suspend fun proper() {
    mutex.withLock {
        doWork()
    } // Lock released even on cancellation
}

// ❌ Risky: Manual lock management
suspend fun risky() {
    mutex.lock()
    try {
        doWork() // If cancelled here, lock stays acquired
    } finally {
        mutex.unlock()
    }
}

性能考量

互斥 vs. synchronized：在协程密集型代码中，互斥更高效，因为线程不会被阻塞
争用：高争用会降低性能；考虑分片（为不同的键设置多个锁）
锁粒度：更细粒度的锁（更多锁，每个锁保护更少的数据）可减少争用
无锁替代方案：对于简单操作，原子类型和 StateFlow 速度更快

示例：分片以减少争用：

class ShardedCache(private val shardCount: Int = 16) {
    private val mutexes = Array(shardCount) { Mutex() }
    private val caches = Array(shardCount) { mutableMapOf<String, Data>() }

    private fun shardIndex(key: String) = key.hashCode() and (shardCount - 1)

    suspend fun put(key: String, value: Data) {
        val index = shardIndex(key)
        mutexes[index].withLock {
            caches[index][key] = value
        }
    }

    suspend fun get(key: String): Data? {
        val index = shardIndex(key)
        return mutexes[index].withLock {
            caches[index][key]
        }
    }
}

真实示例：线程安全的Repository

class UserRepository(
    private val api: UserApi,
    private val database: UserDatabase
) {
    private val cache = mutableMapOf<String, User>()
    private val mutex = Mutex()

    suspend fun getUser(userId: String): User? {
        // Check cache first (read lock)
        mutex.withLock {
            cache[userId]?.let { return it }
        }

        // Try database (outside lock)
        database.getUser(userId)?.let { user ->
            mutex.withLock {
                cache[userId] = user
            }
            return user
        }

        // Fetch from API (outside lock)
        return try {
            val user = api.fetchUser(userId)
            mutex.withLock {
                cache[userId] = user
                database.insertUser(user)
            }
            user
        } catch (e: Exception) {
            null
        }
    }

    suspend fun updateUser(user: User) {
        mutex.withLock {
            cache[user.id] = user
            database.updateUser(user)
        }
    }

    suspend fun clearCache() {
        mutex.withLock {
            cache.clear()
        }
    }
}

测试互斥锁保护的代码

@Test
fun `concurrent increments should be thread-safe`() = runTest {
    val counter = CounterService()

    // Launch 1000 concurrent increments
    val jobs = List(1000) {
        launch {
            counter.increment()
        }
    }

    jobs.joinAll()

    // Should be exactly 1000
    assertEquals(1000, counter.getCount())
}

@Test
fun `mutex prevents race conditions`() = runTest {
    val cache = mutableMapOf<String, Int>()
    val mutex = Mutex()

    // Simulate race condition
    coroutineScope {
        repeat(100) {
            launch {
                mutex.withLock {
                    val current = cache["key"] ?: 0
                    delay(1) // Simulate work
                    cache["key"] = current + 1
                }
            }
        }
    }

    assertEquals(100, cache["key"])
}

总结

Mutex 是一个强大的工具，用于在基于协程的应用程序中保护共享可变状态。它提供线程安全的同步，而不会阻塞线程，使其成为并发协程代码的理想选择。

关键要点：

使用 withLock 进行自动锁管理
保持临界区简洁高效
适当时考虑更简单的替代方案（例如原子操作、StateFlow）
了解何时使用 Mutex 而非其他同步原语
始终妥善处理取消操作

记住：最好的同步就是没有同步。尽可能地，设计系统时，通过使用不可变数据结构、消息传递（Actors/Channels）或响应式流（Flow/StateFlow）来完全避免共享可变状态。但是，当你在协程代码中确实需要互斥时，Mutex 是你的最佳选择。

稀有猿诉

十年磨一剑，历炼出锋芒，说话千百句，不如码二行。