如何同步多线程中的Effect执行，避免Ref读取过期数据

问题场景 #

现有一段使用Ref[F, List[String]]维护状态的Scala并发代码，存在竞态问题：线程读取Ref状态后、执行addToHoldLine修改状态前，其他线程可能读取到过期数据，导致多个线程同时处理同一个key。

原代码：

for
  map <- kafkaEventHoldLine.get // Ref[F, List[String]]
  key = dr.derived + dr.metricId.toString
  _ <- if !map.contains(key) then
    for
      _ <- addToHoldLine(key)
      process <- outlierProcessor.process(dr).timed
      (timeProcess, ol) = process
      _ <- info("Outlier.process", timeProcess.toMillis.named("duration"), dr.derived.named("sid"), dr.metricId.toString.named("mid"))
      append <- ol.traverse_(odt.append).timed
      (timeAppend, _) = append
      _ <- info("Outlier.append", timeAppend.toMillis.named("duration"), dr.derived.named("sid"), dr.metricId.toString.named("mid"))
      _ <- counters.processOutlier.observeTime(timeProcess + timeAppend)
      _ <- clearHoldLine(key)
    yield ()
  else
    Async[F].sleep(delayHoldLine.milliseconds) >> outlierProcess(dr)
yield ()

需要原子同步的核心逻辑：

for
  map <- kafkaEventHoldLine.get // Ref[F, List[String]]
  key = dr.derived + dr.metricId.toString
  _ <- if !map.contains(key) then
    for
      _ <- addToHoldLine(key)

解决方案 #

使用Ref的modify方法原子性完成“检查key是否存在+更新状态”的操作，避免竞态条件。modify会保证整个操作在并发环境下不会被打断，确保状态一致性。

修改后的代码：

for
  key = dr.derived + dr.metricId.toString
  // 原子性检查并更新：key不存在则添加，返回true；存在则返回false
  shouldProcess <- kafkaEventHoldLine.modify { currentList =>
    if currentList.contains(key) then
      (currentList, false)
    else
      (currentList :+ key, true)
  }
  _ <- if shouldProcess then
    for
      process <- outlierProcessor.process(dr).timed
      (timeProcess, ol) = process
      _ <- info("Outlier.process", timeProcess.toMillis.named("duration"), dr.derived.named("sid"), dr.metricId.toString.named("mid"))
      append <- ol.traverse_(odt.append).timed
      (timeAppend, _) = append
      _ <- info("Outlier.append", timeAppend.toMillis.named("duration"), dr.derived.named("sid"), dr.metricId.toString.named("mid"))
      _ <- counters.processOutlier.observeTime(timeProcess + timeAppend)
      // 处理完成后原子性移除key
      _ <- kafkaEventHoldLine.update(_.filterNot(_ == key))
    yield ()
  else
    Async[F].sleep(delayHoldLine.milliseconds) >> outlierProcess(dr)
yield ()

核心要点 #

1. 原子性操作：modify方法将“读取状态-判断-更新状态”合并为一个原子操作，彻底消除竞态窗口。
2. 状态一致性：所有线程看到的Ref状态都是最新的，不会出现读取过期数据的情况。
3. 简化逻辑：移除了单独的addToHoldLine和clearHoldLine封装，直接通过modify和update操作Ref，减少冗余步骤。

内容的提问来源于stack exchange，提问作者Дима Шестаев