FS2 Stream并发优化：依赖单元素流的批量数据入库高效实现

问题根源 #

你用s3.flatMap的方式会导致两个核心瓶颈：

1. s1和s2的处理是串行执行的，大量元素的写入操作会被依次阻塞，无法利用多核资源提升吞吐量
2. 若s1/s2是单条写入数据库，频繁的IO交互会进一步拖慢整体速度

核心优化思路 #

1. 提前获取s3的唯一元素：将s3: Stream[IO, C]转换为IO[C]，只计算一次耗时的s3逻辑
2. 并行处理s1和s2：利用fs2的并行流能力，同时处理两个写入流
3. 批量写入数据库：减少单条写入的IO开销，大幅提升处理效率

代码实现 #

1. 基础并行处理版本

// 定义数据库写入逻辑（需确保是异步IO实现）
def saveA(a: A, c: C): IO[Unit] = ???
def saveB(b: B, c: C): IO[Unit] = ???

// 仅执行一次s3的计算，获取唯一元素
val fetchC: IO[C] = s3.compile.lastOrError

// 并行执行s1和s2的写入流
val optimizedProcessing: IO[Unit] = fetchC.flatMap { c =>
  val processS1 = s1.evalMap(saveA(_, c))
  val processS2 = s2.evalMap(saveB(_, c))
  
  // 并行度根据数据库连接池大小/CPU核心数调整，示例设为2
  fs2.Stream(processS1, processS2).parJoin(2).compile.drain
}

2. 批量写入优化版本（更推荐）

如果数据库支持批量写入，这一步能带来数量级的速度提升：

// 批量写入函数示例
def saveBatchA(as: List[A], c: C): IO[Unit] = ???
def saveBatchB(bs: List[B], c: C): IO[Unit] = ???

val optimizedProcessingWithBatch: IO[Unit] = fetchC.flatMap { c =>
  // 每100个元素为一批（可根据数据库性能调整批次大小）
  val processS1 = s1.chunkN(100).evalMap(chunk => saveBatchA(chunk.toList, c))
  val processS2 = s2.chunkN(100).evalMap(chunk => saveBatchB(chunk.toList, c))
  
  fs2.Stream(processS1, processS2).parJoin(2).compile.drain
}

额外注意事项 #

• 并行度控制：parJoin的参数不要过大，避免数据库连接池耗尽或数据库压力过载，建议参考连接池大小设置（比如连接池有20个连接，并行度设为4-8）
• 异步IO验证：确保数据库操作是异步实现（比如使用fs2-jdbc的异步API），否则线程会被阻塞，并行效果会大幅降低
• 错误处理：可在evalMap后添加handleErrorWith或retry逻辑，处理写入失败的情况，保证数据一致性

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