并发环境下线程安全更新TrieMap中List构建队列的方法

在并发环境下更新TrieMap[A, List[B]]的核心问题是：不可变List的替换操作必须是原子性的，否则多个线程同时读取旧List、修改后写回会覆盖彼此的更新。下面提供两种可靠的解决方案，适配不同的场景需求：

方案一：不可变List + CAS原子替换 #

利用TrieMap的原子replace方法配合循环重试（CAS模式），保证更新操作的原子性。因为List是不可变的，我们每次更新都是创建新List，再原子替换旧值。

代码实现 #

import scala.collection.concurrent.TrieMap

// 初始化TrieMap，存储key到Int队列（用List模拟）的映射
val trieMap: TrieMap[String, List[Int]] = TrieMap()

// 入队操作：原子性添加元素到队列
def enqueue(key: String, elem: Int): Unit = {
  var updated = false
  while (!updated) {
    // 先获取当前队列（不存在则用空列表）
    val currentQueue = trieMap.getOrElse(key, Nil)
    // 用头插法（O(1)效率）构建新队列，后续出队时反转即可得到正确顺序
    val newQueue = elem :: currentQueue
    // 原子替换：只有当前值仍为currentQueue时才替换，避免覆盖其他线程的更新
    updated = trieMap.replace(key, currentQueue, newQueue)
  }
}

// 清空并取出所有元素：原子性替换为空列表，同时返回原队列
def dequeueAll(key: String): List[Int] = {
  var result: List[Int] = Nil
  var done = false
  while (!done) {
    trieMap.get(key) match {
      case Some(currentQueue) =>
        // 原子替换为空列表，确保取出的是当前完整的队列
        if (trieMap.replace(key, currentQueue, Nil)) {
          // 反转头插的列表，恢复队列的先进先出顺序
          result = currentQueue.reverse
          done = true
        }
      case None =>
        done = true // 无元素直接返回空列表
    }
  }
  result
}

优缺点 #

• ✅ 纯Scala不可变集合实现，无外部依赖
• ✅ 入队/清空操作都是原子性的，不会丢失数据
• ❌ 高并发场景下可能因重试导致性能损耗
• ❌ 若用尾插（:+）会导致O(n)时间复杂度，所以必须用头插配合反转

方案二：线程安全可变队列作为Value #

将TrieMap的Value替换为线程安全的可变队列（比如java.util.concurrent.ConcurrentLinkedQueue），利用队列自身的线程安全特性避免竞态条件，性能更优。

代码实现 #

import java.util.concurrent.ConcurrentLinkedQueue
import scala.collection.concurrent.TrieMap

// 初始化TrieMap，存储key到线程安全队列的映射
val trieMap: TrieMap[String, ConcurrentLinkedQueue[Int]] = TrieMap()

// 入队操作：直接调用队列的线程安全add方法
def enqueue(key: String, elem: Int): Unit = {
  // 原子性获取或创建队列（getOrElseUpdate是TrieMap的原子操作）
  val queue = trieMap.getOrElseUpdate(key, new ConcurrentLinkedQueue[Int]())
  queue.add(elem) // add方法本身是线程安全的
}

// 原子性清空并取出所有元素：替换整个队列保证原子性
def dequeueAll(key: String): List[Int] = {
  var result: List[Int] = Nil
  var done = false
  while (!done) {
    trieMap.get(key) match {
      case Some(oldQueue) =>
        // 原子替换为新的空队列，确保后续入队不会干扰当前取出操作
        val newQueue = new ConcurrentLinkedQueue[Int]()
        if (trieMap.replace(key, oldQueue, newQueue)) {
          // 安全遍历旧队列的所有元素
          val buffer = scala.collection.mutable.ListBuffer[Int]()
          var elem = oldQueue.poll()
          while (elem != null) {
            buffer += elem
            elem = oldQueue.poll()
          }
          result = buffer.toList
          done = true
        }
      case None =>
        done = true
    }
  }
  result
}

优缺点 #

• ✅ 入队操作是O(1)时间复杂度，高并发场景下性能更稳定
• ✅ 队列自身线程安全，无需手动CAS循环
• ❌ 依赖Java并发包的队列实现
• ❌ 若直接逐个poll清空，无法保证“原子性取出所有元素”，需配合TrieMap的原子替换实现

关键注意事项 #

• TrieMap本身是线程安全的，但仅保证自身的读写操作原子性；Value层的操作如果不是原子的，仍会出现竞态问题。
• 如果你的Scala版本是2.13+，官方已标记scala.collection.concurrent.TrieMap为废弃，推荐使用java.util.concurrent.ConcurrentHashMap配合Scala的隐式转换（import scala.jdk.CollectionConverters._）来实现类似功能。

内容的提问来源于stack exchange，提问作者Toby