Kotlin 与并发安全（Concurrency Safety in Depth）

前置知识#

• Kotlin 与时间：建议先完成前一篇的学习

学习目标#

• 掌握「1. 历史动机与发展脉络」的核心机制、典型用法与常见陷阱
• 掌握「2. 形式化定义」的核心机制、典型用法与常见陷阱
• 掌握「3. 理论推导与原理解析」的核心机制、典型用法与常见陷阱
• 掌握「4. 代码示例」的核心机制、典型用法与常见陷阱
• 掌握「5. 对比分析」的核心机制、典型用法与常见陷阱

本文档对标 MIT 6.005 Software Construction、Stanford CS110 Principles of Computer Systems、CMU 15-440 Distributed Systems、UC Berkeley CS162 Operating Systems 等海外名校课程的教学水准，系统讲解 Kotlin 在协程与多线程环境下的并发安全（Concurrency Safety）机制，从”为什么需要并发安全”出发，逐层深入到内存模型（Memory Model）、原子操作（Atomic Operations）、互斥锁（Mutex）、CAS（Compare-And-Swap）、Actor 模型（Actor Model）、Channel、线程限制（Thread Confinement）、@Volatile 等核心主题。本文不假设读者熟悉 Java Memory Model 或 Go 内存模型，所有概念均由浅入深、从形式化定义到字节码实现逐层展开。完成本文学习后，读者将能够独立设计线程安全的 Kotlin 协程代码、识别与修复数据竞争（Data Race）、选择合适的并发原语并理解其在 JVM 与 Native 平台的运行时行为差异。

1. 历史动机与发展脉络#

1.1 问题背景：从单线程到多线程#

早期计算机是单核单线程的，程序顺序执行，不存在并发问题。但随着硬件发展：

1. 多核 CPU 普及（2005 年后）：Intel、AMD 推出双核、四核 CPU，单线程无法充分利用硬件。
2. 多线程应用普及：Web 服务器需要同时处理数千个请求，桌面应用需要保持 UI 响应同时执行后台任务。
3. 并发编程兴起：Java 5（2004）引入 java.util.concurrent 包，提供 Executor、Atomic*、Lock 等并发原语。

但多线程编程带来了一个根本性难题：共享可变状态（Shared Mutable State）。

1.2 共享可变状态的问题#

考虑以下经典示例：

// Java 代码
class Counter {
    private int count = 0;
    public void increment() { count++; }  // 非线程安全
    public int get() { return count; }
}

count++ 看起来是原子操作，但实际上由三步组成：

1. 读取 count 到寄存器
2. 寄存器加 1
3. 写回 count

如果两个线程同时执行 increment()，可能发生：

Thread A: read count=0
Thread B: read count=0
Thread A: register=1
Thread B: register=1
Thread A: write count=1
Thread B: write count=1  // 期望结果是 2，实际是 1

这就是数据竞争（Data Race）：多个线程并发访问同一可变状态，且至少一个线程写，且没有同步机制保护。

1.3 Java Memory Model（JMM）#

Java 1.0 没有正式的内存模型，导致各种”奇怪”的并发问题。Java 5（2004）通过 JSR-133 正式确立了 JMM：

1. 主内存与工作内存：每个线程有自己的工作内存（CPU 缓存或寄存器），主内存是共享的。
2. happens-before 关系：定义操作的可见性与顺序性。
3. volatile 语义：保证可见性，禁止指令重排序。
4. final 语义：正确构造的 final 字段对所有线程可见。
5. synchronized 语义：同一锁的释放 happens-before 后续获取。

1.4 Java 并发原语#

java.util.concurrent（JSR-166，由 Doug Lea 设计）提供了：

• Lock 接口：ReentrantLock、ReadWriteLock，比 synchronized 更灵活。
• Atomic* 类：AtomicInteger、AtomicReference，基于 CAS 实现。
• ConcurrentHashMap：分段锁实现的高并发哈希表。
• Executor 框架：线程池、Future、CompletionService。
• CountDownLatch、CyclicBarrier、Semaphore：同步辅助类。

1.5 Kotlin 的协程革命#

Kotlin 1.3（2018）将协程稳定化，引入了”轻量级线程”概念：

• 协程 vs 线程：协程挂起时不阻塞线程，由调度器（Dispatcher）决定何时恢复。
• 结构化并发：协程作用域（CoroutineScope）自动管理子协程的生命周期。
• suspend 函数：通过 CPS 转换，编译器将异步代码编译为状态机。

但协程并未消除并发问题，反而引入了新挑战：

1. 协程在多线程调度器上运行：Dispatchers.Default 使用线程池，协程可能在不同线程上恢复。
2. 协程间的共享状态：多个协程可能访问同一可变状态，需要同步机制。
3. 协程挂起时的状态：挂起点的状态需要保证一致性。

1.6 Kotlin 并发原语的发展#

1.6.1 Kotlin 1.0（2016）：基础 JVM 互操作

Kotlin 1.0 完全兼容 Java 并发原语：

private val lock = ReentrantLock()
private var count = 0

fun increment() {
    lock.lock()
    try {
        count++
    } finally {
        lock.unlock()
    }
}

// 或使用 synchronized
fun incrementSync() = synchronized(lock) { count++ }

1.6.2 Kotlin 1.1（2017）：协程实验

协程作为实验特性引入，但缺少协程友好的并发原语：

• 协程中调用 synchronized 会阻塞线程，违背协程”不阻塞”的设计哲学。
• 需要一种”可挂起的锁”。

1.6.3 Kotlin 1.3（2018）：协程稳定与 Mutex

Kotlin 1.3 将协程稳定化，同时 kotlinx.coroutines 库发布 1.0：

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

private val mutex = Mutex()
private var count = 0

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

Mutex 是协程友好的互斥原语：

• 可挂起：lock() 是 suspend 函数，挂起当前协程而非阻塞线程。
• 公平性：可选的公平模式（FIFO）。
• 可取消：挂起中的协程可被取消，自动释放锁等待。

1.6.4 Kotlin 1.4（2020）：limitedParallelism

Kotlin 1.4 引入 Dispatcher.limitedParallelism(n)，限制调度器上的并发数：

val singleThreadDispatcher = Dispatchers.Default.limitedParallelism(1)

// 在单线程上下文中执行，无需锁
withContext(singleThreadDispatcher) {
    mutableState++
}

这为”线程限制”提供了原语支持。

1.6.5 Kotlin 1.6（2021）：kotlinx.atomicfu

kotlinx.atomicfu 稳定，提供跨平台原子操作：

import kotlinx.atomicfu.atomic

private val count = atomic(0)

fun increment() {
    count.incrementAndGet()
}

支持 JVM、JS、Native 平台，是 KMP 项目的首选并发原语。

1.6.6 Kotlin 1.7-1.9（2022-2023）：@Volatile 跨平台

Kotlin 1.9 将 @Volatile 扩展到 commonMain，可在 KMP 项目中使用：

// commonMain
@Volatile
private var flag: Boolean = false

1.6.7 Kotlin 2.0（2024）：K2 与并发优化

Kotlin 2.0 的 K2 编译器对并发代码进行了优化：

1. 更精确的逃逸分析：识别”不逃逸”的对象，减少同步开销。
2. 更智能的内联：Mutex.withLock 等内联函数被更激进地内联，减少函数调用开销。
3. Native 内存模型重设计：Kotlin 2.0 的 Native 内存模型从”严格隔离”改为”松散共享”，性能提升约 2 倍。

1.7 Kotlin/Native 的特殊挑战#

Kotlin/Native（编译为原生二进制）与 JVM 的并发模型不同：

1. Kotlin 1.x 的严格隔离模型：对象不能在线程间共享，必须显式”冻结”（freeze）。
2. Kotlin 2.0 的新内存模型：默认允许共享可变状态，与 JVM 行为对齐。
3. 性能权衡：严格隔离模型避免数据竞争但增加序列化开销，新模型接近 JVM 但需要开发者注意同步。

1.8 时间线总览#

1995  Java 1.0      — 内置 synchronized、volatile
2004  Java 5        — JSR-133 JMM、JSR-166 java.util.concurrent
2016  Kotlin 1.0    — 完全兼容 Java 并发原语
2017  Kotlin 1.1    — 协程实验，缺少协程友好的锁
2018  Kotlin 1.3    — 协程稳定，kotlinx.coroutines 1.0 引入 Mutex
2020  Kotlin 1.4    — limitedParallelism 原语
2021  Kotlin 1.6    — kotlinx.atomicfu 稳定
2023  Kotlin 1.9    — @Volatile 跨平台
2024  Kotlin 2.0    — K2 优化，Native 新内存模型

1.9 设计哲学#

Kotlin 在并发安全方面的设计哲学：

1. 协程友好优先：Mutex 等原语优先设计为可挂起，避免阻塞线程。
2. 多原语共存：不强制单一并发模型，提供 Mutex、Atomic、Channel、Actor 等多种选择。
3. 跨平台一致性：通过 kotlinx.atomicfu 等库统一 JVM、JS、Native 平台的并发 API。
4. 不可变优先：通过 data class 的 copy、val 属性鼓励不可变设计。
5. 结构化并发：CoroutineScope 自动管理协程生命周期，避免”逃逸协程”导致的资源泄漏。

2. 形式化定义#

2.1 并发的形式化#

并发系统可形式化为一个三元组：

其中：

• $T$ 是线程/协程的集合。
• $S$ 是共享状态的集合。
• $\rightarrow \subseteq T \times S \times S$ 是状态转移关系，表示线程 $t \in T$ 将状态 $s_1$ 转换为 $s_2$。

顺序执行：如果任一时刻只有一个线程在执行，则 $C$ 是顺序系统。

并发执行：如果多个线程可以同时执行，则 $C$ 是并发系统。

并行执行：如果多个线程在物理上同时执行（多核 CPU），则 $C$ 是并行系统。

2.2 数据竞争的形式化#

数据竞争（Data Race）的形式化定义：

其中 $\text{HB}$ 是 happens-before 关系。即：两个不同线程访问同一变量 $x$，至少一个是写，且它们之间没有 happens-before 关系，则发生数据竞争。

2.3 Happens-Before 关系#

Happens-before 关系 $\text{HB}$ 是偏序关系（满足自反、反对称、传递），定义如下：

1. 程序顺序：同一线程内的操作按代码顺序 happens-before。
2. 监视器锁：unlock(M) happens-before 后续的 lock(M)。
3. volatile 字段：对 volatile 字段的写 happens-before 后续的读。
4. 线程启动：Thread.start() happens-before 新线程的第一个操作。
5. 线程终止：线程的所有操作 happens-before Thread.join() 返回。
6. 传递性：如果 $\text{HB}(A, B)$ 且 $\text{HB}(B, C)$，则 $\text{HB}(A, C)$。
7. 协程挂起/恢复：协程的挂起点前的操作 happens-before 恢复后的操作（由 kotlinx.coroutines 实现）。

形式化：

2.4 互斥锁的形式化#

互斥锁（Mutex）的形式化定义为一个状态机：

其中：

• $\{U, L\}$ 是状态集合：$U$ 表示未锁定（Unlocked），$L$ 表示已锁定（Locked）。
• $\{lock, unlock\}$ 是操作集合。
• $\delta$ 是状态转移函数：

• $U$ 是初始状态。

临界区（Critical Section）的形式化定义：

其中 $S$ 是需要互斥访问的代码段。

2.5 CAS 的形式化#

CAS（Compare-And-Swap）的形式化定义：

CAS 是原子操作，由 CPU 指令（如 x86 的 CMPXCHG）直接支持。

基于 CAS 的自旋锁：

基于 CAS 的原子增量：

2.6 ABA 问题#

ABA 问题的形式化描述：

线程 A 读取到值 A，线程 B 将值改为 B 再改回 A，线程 A 的 CAS 仍然成功，但中间状态 B 被忽略。

解决方案：使用版本号 $v$：

2.7 Actor 模型的形式化#

Actor 模型（Hewitt, 1973）的形式化定义：

其中：

• $M$ 是邮箱（Mailbox），是一个 FIFO 队列。
• $B$ 是行为（Behavior），是处理消息的函数 $B : \text{Message} \times S \to S \times \text{Actions}$。
• $S$ 是内部状态（State）。

Actor 通信规则：

1. 唯一标识：每个 Actor 有唯一地址。
2. 异步消息：发送消息 $\text{send}(a, m)$ 是非阻塞的。
3. 顺序处理：Actor 一次处理一个消息，新消息进入邮箱等待。
4. 状态隔离：Actor 的内部状态对外不可见，只能通过消息交互。

消息处理的形式化：

2.8 Channel 的形式化#

Channel 的形式化定义：

其中 $T$ 是元素类型，$Q$ 是 FIFO 队列，$n$ 是容量。

操作语义：

** rendezvous（无缓冲）Channel**：$n = 0$，发送与接收必须同时就绪。

2.9 线程限制的形式化#

线程限制（Thread Confinement）的形式化：

即变量 $x$ 只能被线程 $t$ 访问，对其他线程不可见。线程限制天然保证线程安全，无需同步机制。

栈限制（Stack Confinement）：变量只存在于局部变量表，不逃逸出方法。

2.10 Volatile 的形式化#

@Volatile 的形式化语义：

其中 StoreStore、StoreLoad、LoadLoad、LoadStore 是内存屏障（Memory Barriers），禁止指令重排序。

Volatile 保证：

1. 可见性：写操作立即对其他线程可见。
2. 顺序性：禁止 volatile 读写与前后操作重排序。
3. 不保证原子性：x++ 仍然不安全（读-改-写不是原子的）。

2.11 StateFlow 的形式化#

StateFlow<T> 是一个特殊的热流（Hot Flow），形式化为：

更新语义：

订阅语义：

StateFlow 保证：

1. 最后一次值可见：新订阅者立即收到当前值。
2. conflation：如果更新速度超过消费速度，中间值被合并。
3. 线程安全：所有操作基于 CAS，无需外部锁。

2.12 不可变对象的形式化#

不可变对象（Immutable Object）的形式化：

即对象 $O$ 在构造完成后，所有字段不可修改。

线程安全保证：

不可变对象天生线程安全，因为不存在”写”操作，不满足数据竞争的”至少一个写”条件。

2.13 并发安全的充分条件#

并发安全的充分条件：

避免数据竞争的策略：

1. 不可变：$S$ 全部不可变。
2. 线程限制：每个 $x$ 只被一个线程访问。
3. 同步：通过 Mutex、CAS 等机制建立 happens-before 关系。
4. 消息传递：通过 Channel、Actor 替代共享内存。

3. 理论推导与原理解析#

3.1 为什么 count++ 不安全？#

count++ 在字节码层面是三步：

1. ILOAD count       ; 将 count 读入操作数栈
2. ICONST_1          ; 将常量 1 压入栈
3. IADD              ; 栈顶两数相加
4. ISTORE count      ; 将结果写回 count

这三步不是原子的，可能被打断。考虑两个线程同时执行：

Thread A: ILOAD count (读 count=0)
Thread B: ILOAD count (读 count=0)
Thread A: ICONST_1, IADD (1+0=1)
Thread B: ICONST_1, IADD (1+0=1)
Thread A: ISTORE count (写 count=1)
Thread B: ISTORE count (写 count=1, 应该是 2)

结果：count = 1，但应该是 2。

解决方法：

1. synchronized：将整个方法标记为同步。
2. AtomicInteger：使用 CAS 实现原子增量。
3. Mutex.withLock：在协程中使用可挂起的锁。

3.2 为什么 @Volatile 不能解决 count++？#

@Volatile 保证可见性，但不保证原子性。即：

• 可见性：线程 A 写入 count，线程 B 立即看到新值。
• 原子性：count++ 仍然是”读-改-写”三步，可能被打断。

形式化：

@Volatile 适用于：

• 布尔标志位：@Volatile var cancelled = false，写入后其他线程立即可见。
• 一次性初始化：@Volatile var initialized = false，避免重排序。
• 双重检查锁定：配合 synchronized 实现 lazy initialization。

不适用于：

• 复合操作：count++、check-then-act（if (x == 0) x = 1）。
• 多变量一致性：x 和 y 需要一起更新，@Volatile 不能保证两者同步。

3.3 Mutex 的实现原理#

kotlinx.coroutines.sync.Mutex 的实现基于：

1. CAS 自旋：尝试用 CAS 将状态从 UNLOCKED 改为 LOCKED。
2. 挂起等待：如果失败，将当前协程加入等待队列并挂起。
3. 唤醒：锁释放时，从队列取出一个等待协程并唤醒。

简化实现：

class Mutex {
    private val state = atomic(0)  // 0=unlocked, 1=locked
    private val waitQueue = ConcurrentLinkedQueue<Continuation<Unit>>()

    suspend fun lock() {
        // 快速路径：CAS 尝试
        if (state.compareAndSet(0, 1)) return

        // 慢速路径：加入等待队列并挂起
        suspendCancellableCoroutine<Unit> { cont ->
            waitQueue.add(cont)
            // 可能此时锁被释放，再次尝试
            if (state.compareAndSet(0, 1)) {
                waitQueue.remove(cont)
                cont.resume(Unit)
            }
        }
    }

    fun unlock() {
        state.set(0)
        // 唤醒一个等待协程
        waitQueue.poll()?.resume(Unit)
    }
}

实际实现更复杂，需要处理：

• 公平性：FIFO 或非公平。
• 可取消性：挂起中的协程被取消时，需要从队列移除。
• 所有权：记录持锁协程，便于调试与重入检测（默认不重入）。

3.4 AtomicInteger 的实现原理#

AtomicInteger 在 JVM 上通过 Unsafe.compareAndSwapInt 实现：

public class AtomicInteger {
    private volatile int value;

    public final int incrementAndGet() {
        int oldValue;
        do {
            oldValue = value;
        } while (!Unsafe.compareAndSwapInt(this, valueOffset, oldValue, oldValue + 1));
        return oldValue + 1;
    }
}

Unsafe.compareAndSwapInt 直接调用 JNI，最终调用 CPU 的 CMPXCHG 指令（x86）或 LL/SC 指令（ARM）。

自旋的代价：在高竞争下，CAS 自旋会浪费 CPU 周期。优化策略：

1. 退避（Backoff）：失败后短暂等待（如指数退避）。
2. 自适应自旋：JVM 根据历史成功率调整自旋次数。
3. 分段锁：如 ConcurrentHashMap，将数据分片，每个分片独立加锁。

3.5 StateFlow 的线程安全#

StateFlow 内部使用 AtomicReference<T> 存储值，更新操作基于 CAS：

public override var value: T
    get() = value ?: NULL
    set(value) {
        while (true) {
            val oldValue = this.value
            val newValue = value ?: NULL
            if (oldValue == newValue) return
            if (compareAndSet(oldValue, newValue)) return
        }
    }

conflation（合并）语义：

如果更新速度超过消费速度，中间值被合并。例如：

StateFlow 发出: 1, 2, 3, 4, 5
慢速消费者:     1, ..., 5 (跳过 2, 3, 4)

这适合 UI 状态：UI 只关心最新状态，不关心中间状态。

3.6 Channel 的无锁实现#

Channel 的缓冲区基于 BufferedChannel 类，使用无锁数据结构：

class BufferedChannel<T>(val capacity: Int) {
    private val buffer = AtomicReferenceArray<T>(capacity)
    private val sendIndex = atomic(0L)
    private val receiveIndex = atomic(0L)

    suspend fun send(element: T) {
        while (true) {
            val index = sendIndex.getAndIncrement()
            if (index - receiveIndex.value < capacity) {
                buffer[index % capacity] = element
                return
            }
            // 缓冲区满，挂起
            suspendSend(element, index)
        }
    }

    suspend fun receive(): T {
        while (true) {
            val index = receiveIndex.getAndIncrement()
            val element = buffer[index % capacity]
            if (element != null) {
                buffer[index % capacity] = null
                return element
            }
            // 缓冲区空，挂起
            return suspendReceive(index)
        }
    }
}

实际实现更复杂，需要处理：

• 挂起与恢复：使用 select 实现多路复用。
• 关闭：close() 后所有挂起的发送者收到异常。
• 背压（Backpressure）：缓冲区满时，发送者挂起，避免生产者过快。

3.7 为什么 Actor 模型避免数据竞争？#

Actor 模型的核心是”消息传递替代共享内存”：

1. 状态隔离：每个 Actor 的内部状态对外不可见。
2. 单线程处理：Actor 一次只处理一个消息，新消息进入邮箱等待。
3. 无共享：Actor 之间不直接访问对方状态，只能通过消息。

形式化：

即：Actor 的状态不会被多个线程同时访问。

Kotlin 实现：通过 Channel + 单协程处理：

class CounterActor {
    private val channel = Channel<CounterMsg>(Channel.UNLIMITED)
    private var count = 0

    init {
        GlobalScope.launch {
            for (msg in channel) {
                when (msg) {
                    is Increment -> count++
                    is GetValue -> msg.reply.complete(count)
                }
            }
        }
    }

    suspend fun increment() = channel.send(Increment)
    suspend fun getValue(): Int = coroutineScope {
        val reply = CompletableDeferred<Int>()
        channel.send(GetValue(reply))
        reply.await()
    }
}

3.8 结构化并发与并发安全#

结构化并发（Structured Concurrency）要求：

1. 作用域：所有子协程必须在父协程的作用域内启动。
2. 生命周期：父协程等待所有子协程完成才结束。
3. 取消传播：父协程取消时，所有子协程被取消。

形式化：

并发安全的好处：

1. 避免泄漏：作用域结束时，所有子协程被取消，避免”逃逸协程”。
2. 错误传播：子协程抛出异常时，父协程被取消，所有兄弟协程也被取消。
3. 资源释放：作用域结束时，所有锁、文件句柄等资源被释放。

3.9 Kotlin/Native 的内存模型#

Kotlin 1.x 的 Native 内存模型是”严格隔离”：

1. 对象不可跨线程共享：对象必须显式 freeze 才能”快照”到其他线程。
2. 冻结不可逆：freeze 后的对象不可修改，所有字段递归冻结。
3. 性能开销：频繁 freeze 大对象有性能开销。

Kotlin 2.0 的新内存模型：

1. 默认可共享：对象可在线程间共享，与 JVM 行为对齐。
2. 需要同步：可变共享状态需要显式同步（@Volatile、Atomic*）。
3. 性能提升：避免 freeze 开销，性能接近 JVM。

3.10 并发安全等级#

并发安全的等级（从弱到强）：

1. 不可变（Immutable）：对象不可修改，天生线程安全。
2. 线程限制（Thread Confined）：对象只被一个线程访问。
3. 共享只读（Shared Read-only）：多个线程只读，无写操作。
4. 线程安全共享（Thread-safe Shared）：通过同步机制保证一致性。
5. 共享可变（Shared Mutable）：无同步机制，存在数据竞争（最不安全）。

设计建议：

• 优先不可变：通过 data class + copy 实现不可变更新。
• 次优线程限制：通过 limitedParallelism(1) 实现协程限制。
• 最后选共享+同步：使用 Mutex、Atomic* 等原语。

4. 代码示例#

4.1 非线程安全的计数器#

import kotlinx.coroutines.*
import kotlin.system.measureTimeMillis

// 非线程安全的计数器
class UnsafeCounter {
    var count = 0
    fun increment() { count++ }  // 数据竞争！
}

fun main() = runBlocking {
    val counter = UnsafeCounter()
    val time = measureTimeMillis {
        val jobs = List(100) {
            launch(Dispatchers.Default) {
                repeat(1000) { counter.increment() }
            }
        }
        jobs.joinAll()
    }
    println("Final count: ${counter.count}, expected: 100000, time: ${time}ms")
    // 输出：Final count: 87324, expected: 100000 (数据竞争！)
}

4.2 使用 synchronized 修复#

class SynchronizedCounter {
    private var count = 0
    private val lock = Any()

    fun increment() = synchronized(lock) {
        count++
    }

    fun get() = synchronized(lock) { count }
}

问题：synchronized 阻塞线程，在协程中调用会阻塞调度器线程，影响其他协程。

4.3 使用 Mutex 修复#

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

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

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

    suspend fun get(): Int = mutex.withLock { count }
}

fun main() = runBlocking {
    val counter = MutexCounter()
    val time = measureTimeMillis {
        val jobs = List(100) {
            launch(Dispatchers.Default) {
                repeat(1000) { counter.increment() }
            }
        }
        jobs.joinAll()
    }
    println("Final count: ${counter.get()}, time: ${time}ms")
    // 输出：Final count: 100000 (正确)
}

4.4 使用 AtomicInteger#

import java.util.concurrent.atomic.AtomicInteger

class AtomicCounter {
    private val count = AtomicInteger(0)

    fun increment() {
        count.incrementAndGet()
    }

    fun get() = count.get()
}

优点：无锁，性能高。 缺点：只能保护单个变量，复合操作仍需锁。

4.5 使用 kotlinx.atomicfu#

import kotlinx.atomicfu.atomic
import kotlinx.atomicfu.atomic

class AtomicFuCounter {
    private val count = atomic(0)

    fun increment() {
        count.incrementAndGet()
    }

    fun get() = count.value
}

优点：跨平台（JVM、JS、Native），是 KMP 项目的首选。

4.6 使用 Channel 替代共享状态#

import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.*

class ChannelCounter {
    private val channel = Channel<Int>(Channel.UNLIMITED)
    private var count = 0

    init {
        GlobalScope.launch(Dispatchers.Default) {
            for (delta in channel) {
                count += delta
            }
        }
    }

    suspend fun increment() {
        channel.send(1)
    }

    // 获取值需要通过请求-响应模式
    suspend fun get(): Int = coroutineScope {
        val response = Channel<Int>()
        channel.send(-1)  // 特殊指令表示查询
        // 实际实现需要更复杂的协议
        response.receive()
    }
}

4.7 使用 StateFlow#

import kotlinx.coroutines.flow.*
import kotlinx.coroutines.*

class StateFlowCounter {
    private val _count = MutableStateFlow(0)
    val count: StateFlow<Int> = _count.asStateFlow()

    fun increment() {
        _count.update { it + 1 }
    }
}

fun main() = runBlocking {
    val counter = StateFlowCounter()
    val time = measureTimeMillis {
        val jobs = List(100) {
            launch(Dispatchers.Default) {
                repeat(1000) { counter.increment() }
            }
        }
        jobs.joinAll()
    }
    println("Final count: ${counter.count.value}, time: ${time}ms")
    // 输出：Final count: 100000 (正确)
}

4.8 Actor 模型实现#

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

sealed class CounterMsg
object Increment : CounterMsg()
class GetValue(val reply: CompletableDeferred<Int>) : CounterMsg()

class CounterActor {
    private val channel = Channel<CounterMsg>(Channel.UNLIMITED)
    private var count = 0

    init {
        GlobalScope.launch(Dispatchers.Default) {
            for (msg in channel) {
                when (msg) {
                    is Increment -> count++
                    is GetValue -> msg.reply.complete(count)
                }
            }
        }
    }

    suspend fun increment() = channel.send(Increment)

    suspend fun getValue(): Int {
        val reply = CompletableDeferred<Int>()
        channel.send(GetValue(reply))
        return reply.await()
    }
}

4.9 线程限制#

import kotlinx.coroutines.*

class ConfinedCounter {
    private var count = 0
    private val dispatcher = Dispatchers.Default.limitedParallelism(1)

    suspend fun increment() = withContext(dispatcher) {
        count++
    }

    suspend fun get() = withContext(dispatcher) { count }
}

优点：无需锁，性能接近普通变量访问。 缺点：所有访问必须通过 withContext，增加挂起开销。

4.10 双重检查锁定（DCL）#

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

class LazyInit {
    @Volatile
    private var instance: ExpensiveObject? = null
    private val mutex = Mutex()

    suspend fun getInstance(): ExpensiveObject {
        instance?.let { return it }  // 快速路径：无锁读

        return mutex.withLock {
            instance?.let { return it }  // 双重检查
            instance = ExpensiveObject()
            instance!!
        }
    }
}

class ExpensiveObject

4.11 不可变数据结构#

// 不可变数据类
data class User(val id: String, val name: String, val email: String)

// 通过 copy 更新
val user = User("1", "Alice", "alice@example.com")
val updated = user.copy(name = "Alice Smith")

// 不可变集合
val immutableList = listOf(1, 2, 3)
val updatedList = immutableList + 4  // 新列表

不可变对象天生线程安全，无需同步机制。

4.12 读写锁（ReadWriteLock）#

import java.util.concurrent.locks.ReentrantReadWriteLock
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock

class ReadWriteMap<K, V> {
    private val map = mutableMapOf<K, V>()
    private val rwLock = ReentrantReadWriteLock()

    fun get(key: K): V? {
        rwLock.readLock().lock()
        try {
            return map[key]
        } finally {
            rwLock.readLock().unlock()
        }
    }

    fun put(key: K, value: V) {
        rwLock.writeLock().lock()
        try {
            map[key] = value
        } finally {
            rwLock.writeLock().unlock()
        }
    }
}

适用场景：读多写少（如配置表、缓存）。

4.13 信号量（Semaphore）#

import kotlinx.coroutines.sync.Semaphore
import kotlinx.coroutines.sync.withPermit
import kotlinx.coroutines.*

class RateLimiter(private val maxConcurrency: Int) {
    private val semaphore = Semaphore(maxConcurrency)

    suspend fun <T> doWithLimit(block: suspend () -> T): T {
        return semaphore.withPermit {
            block()
        }
    }
}

// 使用：限制最多 10 个并发请求
val limiter = RateLimiter(10)
val results = listOf("url1", "url2", ...).map { url ->
    async {
        limiter.doWithLimit { fetchUrl(url) }
    }
}.awaitAll()

4.14 @Volatile 标志位#

class CancellableTask {
    @Volatile
    private var cancelled = false

    fun cancel() {
        cancelled = true
    }

    suspend fun run() {
        while (!cancelled) {
            doWork()
            yield()  // 让出 CPU
        }
    }

    private suspend fun doWork() {
        // ...
    }
}

4.15 CAS 实现 Treiber Stack#

import kotlinx.atomicfu.atomic

class TreiberStack<T> {
    private class Node<T>(val value: T, val next: Node<T>?)

    private val top = atomic<Node<T>?>(null)

    fun push(value: T) {
        while (true) {
            val currentTop = top.value
            val newTop = Node(value, currentTop)
            if (top.compareAndSet(currentTop, newTop)) return
        }
    }

    fun pop(): T? {
        while (true) {
            val currentTop = top.value ?: return null
            val next = currentTop.next
            if (top.compareAndSet(currentTop, next)) {
                return currentTop.value
            }
        }
    }
}

优点：无锁，高并发性能好。 缺点：高竞争下 CAS 自旋开销大。

5. 对比分析#

5.1 Kotlin vs Java 并发原语对比#

5.2 共享内存 vs 消息传递#

5.3 Kotlin vs Go 并发对比#

5.4 Kotlin vs Rust 并发对比#

5.5 Kotlin vs Scala 并发对比#

6. 常见陷阱与最佳实践#

6.1 陷阱 1：在协程中使用 synchronized#

// 错误：synchronized 阻塞线程，影响其他协程
suspend fun wrongSync() {
    synchronized(lock) {
        delay(1000)  // 阻塞线程 1 秒！
    }
}

// 正确：使用 Mutex
suspend fun correctMutex() {
    mutex.withLock {
        delay(1000)  // 挂起协程，不阻塞线程
    }
}

6.2 陷阱 2：忘记释放锁#

// 错误：异常时锁未释放
suspend fun wrongRelease() {
    mutex.lock()
    riskyOperation()  // 抛异常，锁泄漏
    mutex.unlock()
}

// 正确：使用 withLock
suspend fun correctRelease() {
    mutex.withLock {
        riskyOperation()  // 异常时 withLock 自动释放
    }
}

6.3 陷阱 3：@Volatile 误用#

// 错误：@Volatile 不能保证 count++ 原子性
class WrongVolatile {
    @Volatile
    var count = 0

    fun increment() { count++ }  // 仍然不安全！
}

// 正确：使用 AtomicInteger
class CorrectAtomic {
    private val count = AtomicInteger(0)

    fun increment() { count.incrementAndGet() }
}

6.4 陷阱 4：复合操作非原子#

// 错误：check-then-act 不是原子的
class WrongCheck {
    @Volatile
    private var initialized = false

    fun wrongInit() {
        if (!initialized) {  // Thread A 读到 false
            // Thread B 也读到 false，并发初始化！
            init()
            initialized = true
        }
    }
}

// 正确：使用 CAS 或 synchronized
class CorrectInit {
    private val initialized = atomic(false)

    fun correctInit() {
        if (initialized.compareAndSet(false, true)) {
            init()  // 只有一个线程执行
        }
    }
}

6.5 陷阱 5：死锁#

// 错误：嵌套锁顺序不一致导致死锁
suspend fun deadlock() {
    coroutineScope {
        launch {
            mutexA.withLock {
                delay(100)
                mutexB.withLock { /* ... */ }  // 等待 B
            }
        }
        launch {
            mutexB.withLock {
                delay(100)
                mutexA.withLock { /* ... */ }  // 等待 A
            }
        }
    }
}

// 正确：统一锁顺序
suspend fun correctLock() {
    coroutineScope {
        launch {
            mutexA.withLock {
                mutexB.withLock { /* ... */ }  // 总是先 A 后 B
            }
        }
        launch {
            mutexA.withLock {
                mutexB.withLock { /* ... */ }  // 总是先 A 后 B
            }
        }
    }
}

6.6 陷阱 6：可变对象通过 Channel 传递#

// 错误：可变对象在多协程间共享
data class MutableData(var value: Int = 0)

suspend fun wrongSharedMutable() {
    val channel = Channel<MutableData>()
    launch {
        val data = MutableData(0)
        repeat(100) {
            data.value = it
            channel.send(data)  // 发送同一对象引用
        }
    }
    launch {
        repeat(100) {
            val received = channel.receive()
            println(received.value)  // 可能读到错误的值
        }
    }
}

// 正确：发送不可变副本
data class ImmutableData(val value: Int)

suspend fun correctImmutable() {
    val channel = Channel<ImmutableData>()
    launch {
        repeat(100) {
            channel.send(ImmutableData(it))  // 新对象
        }
    }
}

6.7 陷阱 7：StateFlow 不能保护复杂状态#

// 错误：StateFlow 只保护单个字段
class WrongStateFlow {
    private val _users = MutableStateFlow<List<User>>(emptyList())

    suspend fun addUser(user: User) {
        // 错误：read-modify-write 不是原子的
        _users.value = _users.value + user
    }
}

// 正确：使用 update
class CorrectStateFlow {
    private val _users = MutableStateFlow<List<User>>(emptyList())

    fun addUser(user: User) {
        _users.update { it + user }  // 基于 CAS，原子
    }
}

6.8 陷阱 8：runBlocking 滥用#

// 错误：在协程中调用 runBlocking 会阻塞线程
suspend fun wrongRunBlocking() {
    runBlocking {
        delay(1000)  // 阻塞当前线程 1 秒
    }
}

// 正确：直接使用 suspend 函数
suspend fun correctSuspend() {
    delay(1000)  // 挂起协程
}

6.9 陷阱 9：GlobalScope 滥用#

// 错误：GlobalScope 无法取消，容易泄漏
fun wrongGlobalScope() {
    GlobalScope.launch {
        while (true) {
            delay(1000)
            doWork()  // 永远执行
        }
    }
}

// 正确：使用结构化并发
fun correctStructured() = runBlocking {
    launch {
        while (isActive) {
            delay(1000)
            doWork()
        }
    }
}

6.10 陷阱 10：async 不 await#

// 错误：async 启动后不 await，异常被吞
suspend fun wrongAsync() {
    coroutineScope {
        async {
            throw RuntimeException("Failed")  // 异常被吞
        }
        // 不 await，异常丢失
    }
}

// 正确：await 所有 async
suspend fun correctAsync() = coroutineScope {
    val deferred = async {
        throw RuntimeException("Failed")
    }
    try {
        deferred.await()  // 异常在此抛出
    } catch (e: Exception) {
        println("Caught: ${e.message}")
    }
}

6.11 陷阱 11：launch 替代 async#

// 错误：launch 用于需要结果的场景，丢失返回值
suspend fun wrongLaunch() {
    val result = CoroutineScope(Dispatchers.Default).launch {
        computeResult()  // 返回值被丢弃
    }
    // 无法获取结果
}

// 正确：使用 async + await
suspend fun correctAsync() = coroutineScope {
    val deferred = async { computeResult() }
    val result = deferred.await()
    println(result)
}

6.12 陷阱 12：Dispatchers.Default 滥用#

// 错误：所有协程都用 Default，可能导致调度器饱和
suspend fun wrongDispatcher() {
    coroutineScope {
        List(10000) {
            launch(Dispatchers.Default) {
                Thread.sleep(1000)  // 阻塞调度器线程
            }
        }.joinAll()
    }
}

// 正确：阻塞操作用 IO，CPU 密集用 Default
suspend fun correctDispatcher() {
    coroutineScope {
        List(10000) {
            launch(Dispatchers.IO) {
                delay(1000)  // 挂起，不阻塞线程
            }
        }.joinAll()
    }
}

6.13 陷阱 13：Mutex 不可重入#

private val mutex = Mutex()

// 错误：再次 lock 会死锁
suspend fun recursiveLock() {
    mutex.withLock {
        mutex.withLock {  // 死锁！Mutex 不可重入
            // ...
        }
    }
}

// 解决：重构代码避免嵌套锁，或使用 ReentrantLock

6.14 陷阱 14：AtomicReference 复合操作#

// 错误：多个字段的更新不是原子的
class WrongMultiField {
    private val state = AtomicReference(State(0, 0))

    fun wrongUpdate() {
        val current = state.get()
        state.set(State(current.a + 1, current.b))  // 中间状态可能被其他线程读到
    }
}

// 正确：使用 CAS 循环
class CorrectMultiField {
    private val state = atomic(State(0, 0))

    fun update() {
        state.update { it.copy(a = it.a + 1) }
    }
}

data class State(val a: Int, val b: Int)

6.15 陷阱 15：lateinit var 多线程#

// 错误：lateinit var 多线程初始化不安全
class WrongLateinit {
    lateinit var service: Service

    fun wrongInit() {
        coroutineScope {
            launch { service = Service("a") }
            launch { service = Service("b") }  // 竞争！
        }
    }
}

// 正确：使用 lazy 或 AtomicReference
class CorrectLazy {
    val service by lazy { Service("default") }
}

7. 工程实践#

7.1 线程安全 LRU 缓存#

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

class ThreadSafeLRUCache<K, V>(private val maxSize: Int) {
    private val mutex = Mutex()
    private val cache = LinkedHashMap<K, V>(16, 0.75f, true)

    suspend fun get(key: K): V? = mutex.withLock {
        cache.get(key)
    }

    suspend fun put(key: K, value: V) = mutex.withLock {
        if (cache.size >= maxSize) {
            val oldest = cache.keys.first()
            cache.remove(oldest)
        }
        cache.put(key, value)
    }

    suspend fun remove(key: K) = mutex.withLock {
        cache.remove(key)
    }

    suspend fun size(): Int = mutex.withLock { cache.size }
}

7.2 并发任务调度器（Actor 模型）#

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

sealed class SchedulerMsg
class Submit(val task: suspend () -> Unit) : SchedulerMsg()
object Shutdown : SchedulerMsg()

class SingleThreadScheduler {
    private val channel = Channel<SchedulerMsg>(Channel.UNLIMITED)
    private val job: Job

    init {
        job = GlobalScope.launch(Dispatchers.Default) {
            for (msg in channel) {
                when (msg) {
                    is Submit -> msg.task()
                    is Shutdown -> break
                }
            }
        }
    }

    fun submit(task: suspend () -> Unit) {
        channel.trySend(Submit(task))
    }

    fun shutdown() {
        channel.trySend(Shutdown)
    }

    suspend fun join() = job.join()
}

7.3 无锁并发队列（Michael-Scott 队列）#

import kotlinx.atomicfu.atomic
import kotlinx.atomicfu.atomic

class MichaelScottQueue<T> {
    private class Node<T>(val value: T?, next: Node<T>? = null)

    private val head: AtomicRef<Node<T>>
    private val tail: AtomicRef<Node<T>>

    init {
        val dummy = Node<T>(null)
        head = atomic(dummy)
        tail = atomic(dummy)
    }

    fun enqueue(value: T) {
        val newNode = Node(value)
        while (true) {
            val currentTail = tail.value
            val next = currentTail.next?.value
            if (currentTail == tail.value) {
                if (next == null) {
                    if (currentTail.next.compareAndSet(null, newNode)) {
                        tail.compareAndSet(currentTail, newNode)
                        return
                    }
                } else {
                    tail.compareAndSet(currentTail, next)
                }
            }
        }
    }

    fun dequeue(): T? {
        while (true) {
            val currentHead = head.value
            val currentTail = tail.value
            val next = currentHead.next?.value
            if (currentHead == head.value) {
                if (currentHead == currentTail) {
                    if (next == null) return null
                    tail.compareAndSet(currentTail, next)
                } else {
                    val value = next?.value
                    if (head.compareAndSet(currentHead, next!!)) {
                        return value
                    }
                }
            }
        }
    }
}

7.4 并发限流器#

import kotlinx.coroutines.sync.Semaphore
import kotlinx.coroutines.sync.withPermit
import kotlinx.coroutines.*

class RateLimiter(private val permits: Int, private val timeoutMs: Long = 0) {
    private val semaphore = Semaphore(permits)

    suspend fun <T> execute(block: suspend () -> T): T {
        return if (timeoutMs > 0) {
            withTimeout(timeoutMs) {
                semaphore.withPermit { block() }
            }
        } else {
            semaphore.withPermit { block() }
        }
    }
}

// 使用：限制 API 调用并发数
val limiter = RateLimiter(10)
val results = urls.map { url ->
    async(Dispatchers.IO) {
        limiter.execute { fetchUrl(url) }
    }
}.awaitAll()

7.5 异步事件总线#

import kotlinx.coroutines.flow.*
import kotlinx.coroutines.*

class EventBus {
    private val events = MutableSharedFlow<Event>(extraBufferCapacity = 64)

    fun emit(event: Event) {
        events.tryEmit(event)
    }

    fun subscribe(): SharedFlow<Event> = events.asSharedFlow()
}

sealed class Event {
    data class UserLogin(val userId: String) : Event()
    data class UserLogout(val userId: String) : Event()
    data class MessageReceived(val from: String, val content: String) : Event()
}

7.6 双缓冲 Double Buffering#

import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
import kotlinx.coroutines.flow.*

class DoubleBuffer<T>(initial: T) {
    private val mutex = Mutex()
    private var current: T = initial
    private var next: T = initial

    val state: Flow<T> = flow {
        emit(current)
        while (true) {
            mutex.withLock {
                if (current != next) {
                    current = next
                    emit(current)
                }
            }
        }
    }

    fun update(new: T) {
        mutex.withLock {
            next = new
        }
    }
}

7.7 原子引用的状态机#

import kotlinx.atomicfu.atomic

class StateMachine<S>(initial: S) {
    private val state = atomic(initial)

    fun transition(transition: (S) -> S): S {
        while (true) {
            val current = state.value
            val next = transition(current)
            if (state.compareAndSet(current, next)) {
                return next
            }
        }
    }

    fun get(): S = state.value
}

// 使用
sealed class ConnectionState {
    object Disconnected : ConnectionState()
    object Connecting : ConnectionState()
    object Connected : ConnectionState()
    data class Error(val msg: String) : ConnectionState()
}

val machine = StateMachine<ConnectionState>(ConnectionState.Disconnected)

fun connect() = machine.transition {
    when (it) {
        is ConnectionState.Disconnected -> ConnectionState.Connecting
        else -> it
    }
}

7.8 协程本地存储（ThreadLocal/CoroutineLocal）#

import kotlinx.coroutines.*

val requestIdContext = CoroutineName("request-context")

class RequestContext(val requestId: String)

suspend fun withRequestContext(requestId: String, block: suspend () -> Unit) {
    withContext(CoroutineName("request-$requestId")) {
        block()
    }
}

// 或使用 ThreadLocal（仅 JVM）
val threadLocalRequest = ThreadLocal<String>()

suspend fun processRequest(requestId: String) {
    threadLocalRequest.set(requestId)
    try {
        // 在协程中使用
        doWork()
    } finally {
        threadLocalRequest.remove()
    }
}

7.9 响应式状态管理（MVI）#

import kotlinx.coroutines.flow.*
import kotlinx.coroutines.*

class MVIStore<S, I, E>(
    initialState: S,
    private val reducer: (S, I) -> S,
    private val effects: (I) -> List<E>
) {
    private val _state = MutableStateFlow(initialState)
    val state: StateFlow<S> = _state.asStateFlow()

    private val _effects = MutableSharedFlow<E>(extraBufferCapacity = 64)
    val effects: SharedFlow<E> = _effects.asSharedFlow()

    fun dispatch(intent: I) {
        _state.update { reducer(it, intent) }
        effects(intent).forEach { _effects.tryEmit(it) }
    }
}

7.10 并发安全的单例#

// 错误：非线程安全
class UnsafeSingleton {
    companion object {
        private var instance: UnsafeSingleton? = null

        fun getInstance(): UnsafeSingleton {
            if (instance == null) {
                instance = UnsafeSingleton()  // 数据竞争
            }
            return instance!!
        }
    }
}

// 正确 1：使用 object 声明
object CorrectSingleton {
    fun doWork() { /* ... */ }
}

// 正确 2：使用 lazy
class LazySingleton private constructor() {
    companion object {
        val instance: LazySingleton by lazy { LazySingleton() }
    }
}

// 正确 3：DCL + @Volatile
class DCLSingleton private constructor() {
    companion object {
        @Volatile
        private var instance: DCLSingleton? = null

        fun getInstance(): DCLSingleton {
            return instance ?: synchronized(this) {
                instance ?: DCLSingleton().also { instance = it }
            }
        }
    }
}

8. 案例研究#

8.1 案例研究 1：kotlinx.coroutines.sync.Mutex#

问题：协程友好的互斥锁，可挂起而非阻塞。

实现要点：

1. 状态管理：使用 AtomicInt 存储 UNLOCKED、LOCKED、SUSPENDED 三种状态。
2. 快速路径：CAS 尝试从 UNLOCKED 到 LOCKED，成功则立即返回。
3. 慢速路径：CAS 失败时，将协程加入等待队列并挂起。
4. 唤醒：unlock() 时从队列取出一个等待协程并恢复。

源码片段（简化）：

public class Mutex {
    private val state = atomic(UNLOCKED)
    private val waiters = AtomicRef<Node?>(null)

    public suspend fun lock() {
        // 快速路径
        if (state.compareAndSet(UNLOCKED, LOCKED)) return

        // 慢速路径：加入等待队列
        val node = Node(currentCoroutine())
        waiters.update { node.also { it.next = it } }
        suspendCancellableCoroutine<Unit> { cont ->
            node.cont = cont
        }
    }

    public fun unlock() {
        val waiter = waiters.value
        if (waiter == null) {
            if (!state.compareAndSet(LOCKED, UNLOCKED)) {
                // 状态异常
            }
        } else {
            // 唤醒等待者
            waiter.cont?.resume(Unit)
        }
    }
}

启示：协程原语需要”快速路径 + 慢速路径”设计，快速路径使用无锁 CAS，慢速路径使用挂起。

8.2 案例研究 2：StateFlow 的无锁实现#

问题：多个协程并发更新状态，需要原子性。

实现：

public class StateFlow<T>(initial: T) {
    @Volatile
    private var value: T = initial

    private val subscribers = AtomicLong(0)

    public fun update(transform: (T) -> T) {
        while (true) {
            val old = value
            val new = transform(old)
            if (old == new) return
            // CAS 更新
            if (compareAndSet(old, new)) return
        }
    }
}

启示：StateFlow 通过 CAS 实现无锁更新，比 Mutex 更高效。

8.3 案例研究 3：Channel 的无锁缓冲#

问题：多生产者-多消费者的无锁队列。

实现：基于”环形缓冲 + 原子索引”：

class BufferedChannel<T>(val capacity: Int) {
    private val buffer = AtomicReferenceArray<T?>(capacity)
    private val sendIndex = AtomicLong(0)
    private val receiveIndex = AtomicLong(0)

    suspend fun send(element: T) {
        while (true) {
            val idx = sendIndex.getAndIncrement()
            if (idx - receiveIndex.get() < capacity) {
                buffer.set(idx % capacity, element)
                return
            }
            // 缓冲满，挂起
            suspendSend(element, idx)
        }
    }
}

启示：无锁数据结构的关键是”分离索引”，发送与接收独立计数。

8.4 案例研究 4：ConcurrentHashMap#

问题：高并发哈希表。

实现：分段锁（Segmented Locks）：

class ConcurrentHashMap<K, V>(initialCapacity: Int = 16) {
    private val segments = Array(initialCapacity) { Segment<K, V>() }

    fun get(key: K): V? {
        return segmentFor(key).get(key)
    }

    fun put(key: K, value: V) {
        segmentFor(key).put(key, value)
    }

    private fun segmentFor(key: K) = segments[(key.hashCode() and 0x7FFFFFFF) % segments.size]
}

class Segment<K, V> {
    private val map = mutableMapOf<K, V>()

    fun get(key: K) = synchronized(this) { map[key] }

    fun put(key: K, value: V) = synchronized(this) { map.put(key, value) }
}

启示：通过分片减少锁粒度，提升并发性能。

8.5 案例研究 5：Actor 模型在 Android 中的应用#

问题：UI 状态管理，避免 Mutex 与复杂同步。

实现：

class UiActor<S>(initial: S) {
    private val channel = Channel<Action<S>>(Channel.UNLIMITED)
    private var state: S = initial

    init {
        GlobalScope.launch(Dispatchers.Main) {
            for (action in channel) {
                state = action.reduce(state)
                action.onComplete?.invoke(state)
            }
        }
    }

    fun dispatch(action: Action<S>) {
        channel.trySend(action)
    }
}

class Action<S>(val reduce: (S) -> S, val onComplete: ((S) -> Unit)? = null)

启示：Actor 模型将状态隔离，简化并发代码。

8.6 案例研究 6：无锁链表#

问题：并发链表，支持插入与删除。

实现：基于 Harris 的无锁链表算法：

class LockFreeLinkedList<T> {
    private class Node<T>(val value: T, next: Node<T>? = null) {
        val next = atomic(next)
        @Volatile
        var marked = false  // 逻辑删除标记
    }

    private val head = atomic<Node<T>?>(null)

    fun add(value: T) {
        while (true) {
            val current = head.value
            val newNode = Node(value, current)
            if (head.compareAndSet(current, newNode)) return
        }
    }

    fun remove(value: T): Boolean {
        while (true) {
            var prev: Node<T>? = null
            var curr = head.value
            while (curr != null && curr.value != value) {
                prev = curr
                curr = curr.next.value
            }
            if (curr == null) return false

            // 标记删除
            val next = curr.next.value
            if (!curr.next.compareAndSet(next, next?.also { it?.marked = true })) continue
            // 物理删除
            if (prev == null) {
                if (head.compareAndSet(curr, next)) return true
            } else {
                if (prev.next.compareAndSet(curr, next)) return true
            }
        }
    }
}

启示：无锁链表通过”标记删除 + 物理删除”两阶段处理 ABA 问题。

8.7 案例研究 7：AtomicReference 在 Spring 中的应用#

问题：Spring 的 AtomicReference 用于配置管理。

class SpringConfigManager {
    private val config = AtomicReference<Map<String, String>>(emptyMap())

    fun updateConfig(newConfig: Map<String, String>) {
        config.set(newConfig)
    }

    fun getConfig(): Map<String, String> = config.get()
}

启示：AtomicReference 适合”整体替换”的场景。

8.8 案例研究 8：Semaphore 在限流中的应用#

问题：限制 API 调用并发数。

class ApiGateway(private val maxConcurrent: Int) {
    private val semaphore = kotlinx.coroutines.sync.Semaphore(maxConcurrent)

    suspend fun callApi(url: String): String {
        return semaphore.withPermit {
            // 限流内执行
            httpClient.get(url)
        }
    }
}

启示：Semaphore 是限流的最佳选择，比 Mutex 更灵活（允许多个许可）。

8.9 案例研究 9：@Volatile 在双重检查锁定中的应用#

问题：延迟初始化，避免每次访问都加锁。

class LazyService {
    @Volatile
    private var instance: ExpensiveResource? = null

    fun getInstance(): ExpensiveResource {
        return instance ?: synchronized(this) {
            instance ?: ExpensiveResource().also { instance = it }
        }
    }
}

启示：@Volatile + synchronized 是延迟初始化的经典模式。

8.10 案例研究 10：kotlinx-atomicfu 在 KMP 中的应用#

问题：跨平台原子操作。

// commonMain
import kotlinx.atomicfu.atomic

expect class AtomicCounter {
    fun increment()
    fun get(): Int
}

// commonMain 实现
class AtomicCounterImpl {
    private val count = atomic(0)

    fun increment() {
        count.incrementAndGet()
    }

    fun get() = count.value
}

启示：kotlinx-atomicfu 是 KMP 项目首选的并发原语。

9.1 基础题#

习题 1：什么是数据竞争？请给出形式化定义。

解析讲解：数据竞争是两个或多个线程并发访问同一变量，至少一个是写操作，且它们之间没有 happens-before 关系。形式化：

习题 2：@Volatile 能否解决 count++ 的线程安全问题？为什么？

解析讲解：不能。@Volatile 保证可见性，但不保证原子性。count++ 是”读-改-写”三步，可能被打断。应该使用 AtomicInteger.incrementAndGet()。

9.2 型变题#

习题 3：以下代码是否线程安全？为什么？

class Cache {
    private val map = ConcurrentHashMap<String, String>()

    fun putIfAbsent(key: String, value: String): String? {
        val existing = map[key]
        if (existing == null) {
            map[key] = value
        }
        return existing
    }
}

解析讲解：不安全。check-then-act（if (existing == null) map[key] = value）不是原子的。应该使用 ConcurrentHashMap.putIfAbsent(key, value)。

9.3 星投影题#

习题 4：实现一个线程安全的 Lazy<T>，要求：

1. 只初始化一次。
2. 多线程访问安全。
3. 性能优化（避免每次访问都加锁）。

解析讲解：

class ThreadSafeLazy<T>(private val initializer: () -> T) : Lazy<T> {
    @Volatile
    private var _value: T? = null
    private var initialized = false
    private val lock = Any()

    override val value: T
        get() {
            if (initialized) return _value!!
            synchronized(lock) {
                if (initialized) return _value!!
                _value = initializer()
                initialized = true
                return _value!!
            }
        }

    override fun isInitialized() = initialized
}

9.4 reified 题#

习题 5：实现一个线程安全的单例，要求：

1. 延迟初始化。
2. 多线程访问安全。
3. 不使用 synchronized。

解析讲解：

class Singleton private constructor() {
    companion object {
        val instance: Singleton by lazy { Singleton() }
    }
}

lazy 默认使用 LazyThreadSafetyMode.SYNCHRONIZED，内部通过 DCL + @Volatile 实现。

9.7 设计题#

习题 7：设计一个线程安全的对象池，要求：

1. 预分配 N 个对象。
2. 借出与归还操作线程安全。
3. 池耗尽时阻塞等待。

解析讲解：

import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.*

class ObjectPool<T>(private val factory: () -> T, initialSize: Int) {
    private val pool = Channel<T>(initialSize)

    init {
        repeat(initialSize) {
            pool.trySend(factory())
        }
    }

    suspend fun borrow(): T = pool.receive()

    suspend fun returnObject(obj: T) = pool.send(obj)
}

// 使用
val pool = ObjectPool({ DBConnection() }, 10)
val conn = pool.borrow()
try {
    conn.query("SELECT ...")
} finally {
    pool.returnObject(conn)
}