6.3 并发集合与线程安全

6.3 并发集合与线程安全

并发集合解决"多个线程确实要碰同一份数据"的场景：TrieMap 提供无锁快照读，Java 的并发队列提供跨线程交接。本节同时给出"哪些集合在并发下不安全"的清单与排查思路。

先划红线：默认集合不线程安全

val seen = mutable.HashMap.empty[String, Int]
// 多线程同时执行：
seen(word) = seen.getOrElse(word, 0) + 1
// 后果：丢更新、读到半构建状态、偶发死循环 —— 全都难复现
​

读写的复合操作不是原子的，即使 HashMap 内部某一步是。并发 bug 的特征是压测偶现、单测全绿，所以红线要划在评审阶段而不是复现之后。

TrieMap：并发读多写少

import scala.collection.concurrent.TrieMap

val cache = TrieMap.empty[String, BigDecimal]
cache("BTC") = price          // 并发安全
cache("BTC")                  // 无锁读

val snapshot: Map[String, BigDecimal] = cache.snapshot  // 一致性快照
​

TrieMap（无锁 CAS 前缀树）读完全无锁、写细粒度并发，还能 O(1) 拿到一致性快照——适合配置表、路由表这类读远多于写的共享结构。

跨线程交接任务用 Java 生态的队列（7.2 的互操作在此提前亮相）：

val queue = java.util.concurrent.LinkedBlockingQueue[String](1000)
queue.put(task)        // 满则阻塞 —— 天然背压
val t = queue.poll(1, TimeUnit.SECONDS)
​

并发数据工具的选型地图

原子变量与不可变的联防

简单计数不必上重型结构，AtomicLong 或 6.1 的 Future.sequence 汇总都更轻：

val hits = java.util.concurrent.atomic.AtomicLong()
hits.incrementAndGet()

// 函数式替代：各线程算各自的，最后合并不可变结果
val counts: List[Map[String, Int]] = pages.map(countAsync).sequence.map(_.Await)
val merged = counts.reduce(_ ++ _ withMerge)
​

优先让线程不共享（各算各的最后合并），其次选并发原语，最后才考虑共享可变结构——顺序与 3.1 的不可变哲学一脉相承。

图 6-1 并发安全的三道防线

竞态现场还原与三种修法

先看一段必然出错的代码，再对比三种修复路线：

// 事故现场：检查与写入之间有空窗，两个线程同时通过检查
var seen = mutable.Map.empty[String, Int]
def bump(k: String): Unit =
  if !seen.contains(k) then seen(k) = 1
  else seen(k) = seen(k) + 1        // 并发下丢更新

// 修法一：原子引用 + 不可变 Map 的 CAS 循环
val ref = AtomicReference(Map.empty[String, Int])
def bump1(k: String): Unit =
  var continue = true
  while continue do
    val cur = ref.get
    val nxt = cur.updated(k, cur.getOrElse(k, 0) + 1)
    continue = !ref.compareAndSet(cur, nxt)

// 修法二：并发 TrieMap，单键操作内部已并发安全
val triei = new concurrent.TrieMap[String, Int]()
def bump2(k: String): Unit = triei.update(k, triei.getOrElse(k, 0) + 1)
​

修法一无锁但代码啰嗦，适合读多写少；修法二一行搞定，适合通用场景；修法三回到 Actor（6.2），把计数收进单点。三条路线的选择标准是状态复杂度：一两个字段用原子引用，一个集合用 TrieMap，一组相关状态用 Actor。

概念辨析：线程安全集合不等于线程安全程序

val q = new java.util.concurrent.ConcurrentLinkedQueue[Int]()
q.add(1)
if q.isEmpty then q.add(2)   // isEmpty 与 add 各自安全，组合仍有空窗
​

并发集合保证单个方法原子，不保证多个方法的序列原子。需要复合操作原子时，要么用集合自带的复合原语，要么显式加同步块——这条辨析是并发面试与事故复盘的共同高频考点。

java.util.concurrent 借用指南

Scala 并发代码经常直接复用 JVM 成熟组件，三个最常用：ConcurrentHashMap 的 computeIfAbsent 做"不存在则计算"的一次性建键；BlockingQueue 做生产者消费者队列，put/take 自带阻塞语义；CountDownLatch 做一次性栅栏，主线程 await 到全部工作完成。一个快速鉴别法收尾：看到 synchronized 关键字先问保护的是哪个不变量，答不上来的锁多半迟早出事；看到 TrieMap 则确认出口是否 toMap 冻结——两种集合两种纪律，考核时也常被当作判断题来问，答案永远是那句：单操作原子，复合不原子。它们与 TrieMap 的分工：跨语言团队共享代码或需要阻塞语义时用 JDK 组件，纯 Scala 侧优先 TrieMap 以保持不可变出口的一致性。

本节要点回顾

• 默认可变集合并发即雷，复合读写不原子，bug 压测才现形。
• TrieMap 读无锁、可快照，读多写少标配；BlockingQueue 负责交接且有界背压。
• 三级防线：不共享 → 原子变量/合并 → 并发集合。
• 写密集复杂状态时，答案是 6.2 的 Actor 而不是更强的集合。