基于Scala Cats与FS2 parEvalMap的竞态条件解决方案咨询

针对你用FS2 parEvalMap(20)处理值流时遇到的读写竞态问题，以下是几个Cats Effect/FS2原生内置的解决方案：

1. 利用Cats Effect Mutex实现互斥访问 #

这是最通用的同步方案，适用于任意读写逻辑的互斥控制，完全内置在Cats Effect中，无需额外依赖。通过Mutex的lock.surround方法，确保同一时间只有一个线程执行被包裹的读写操作：

import cats.effect.Mutex
import cats.effect.IO
import fs2.Stream

// 初始化Mutex（建议通过Resource或在IO上下文内初始化）
val mutex: IO[Mutex[IO]] = Mutex[IO]

// 包装读写操作
def safeRead: IO[YourValueType] = mutex.flatMap(_.lock.surround(yourReadOperation))
def safeWrite(newValue: YourValueType): IO[Unit] = mutex.flatMap(_.lock.surround(yourWriteOperation(newValue)))

// 在parEvalMap中使用安全的读写方法
val processedStream: Stream[IO, Unit] = yourStream.parEvalMap(20) { value =>
  if (isReadOperation(value)) safeRead.void
  else safeWrite(extractValue(value))
}

2. 用Cats Effect Ref管理原子状态 #

如果你的读写操作是针对某个共享状态的，Ref是更轻量的选择。Ref的所有操作（get/set/modify）都是原子性的，天生线程安全，无需额外同步：

import cats.effect.Ref
import cats.effect.IO
import fs2.Stream

// 初始化共享状态（同样建议在IO/Resource内初始化）
val sharedState: IO[Ref[IO, YourValueType]] = Ref.initial[IO, YourValueType]

// 基于Ref的安全读写
def safeRead: IO[YourValueType] = sharedState.flatMap(_.get)
def safeWrite(newValue: YourValueType): IO[Unit] = sharedState.flatMap(_.set(newValue))

// 复杂的原子更新可以用modify
def updateState(updateFn: YourValueType => YourValueType): IO[YourValueType] = 
  sharedState.flatMap(_.modify(current => (updateFn(current), current)))

在parEvalMap中直接调用这些方法即可，Ref会自动保证操作的原子性，避免竞态。

3. 切换为串行处理（简单但牺牲性能） #

如果业务允许放弃并行处理，直接用FS2的evalMap替代parEvalMap，所有读写操作会串行执行，从根本上消除竞态：

val processedStream: Stream[IO, Unit] = yourStream.evalMap { value =>
  if (isReadOperation(value)) yourReadOperation.void
  else yourWriteOperation(extractValue(value))
}

这种方案最简单，但会失去parEvalMap带来的并行性能提升，仅适合对性能要求不高的场景。

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