基于Scala IOApp实现100个Fiber并发递增Ref并计时需求

以下是满足需求的完整Scala代码，基于Cats Effect实现：

import cats.effect.{IO, IOApp, Ref}
import scala.concurrent.duration.FiniteDuration

object Fibers extends IOApp.Simple {

  private val iterationsPerFiber = 1000
  private val fiberCount = 100

  val program =
    for {
      ref <- Ref.of[IO, Int](0)
      // 定义单个Fiber的任务：完成1000次Ref递增
      fiberTask = IO.iterate(0)(_ + 1).take(iterationsPerFiber).flatMap(_ => ref.update(_ + 1)).void
      // 启动100个Fiber并等待全部完成，同时记录耗时
      (duration, _) <- List.fill(fiberCount)(fiberTask).parSequence.void.timed
      total <- ref.get
      _ <- IO.println(s"ref total is: $total")
      _ <- IO.println(s"操作耗时: ${duration.toMillis} 毫秒")
    } yield ()

  override def run: IO[Unit] = program
}

关键细节说明 #

• 共享状态安全：使用Ref作为线程安全的共享状态容器，它的update方法是原子操作，避免了多Fiber并发修改时的竞态问题，这是Cats Effect中处理可变共享状态的推荐方案。
• 单个Fiber逻辑：每个Fiber通过IO.iterate生成1000次迭代，每次迭代执行一次Ref递增，确保每个Fiber严格完成1000次操作后结束。
• 并发控制：List.fill(fiberCount)(fiberTask)创建100个任务实例，parSequence会将这些任务并发执行，并等待所有Fiber执行完成后才继续后续流程，保证所有递增操作都完成后再读取最终值。
• 耗时统计：timed方法包裹并发任务块，返回操作的总执行时长，我们将其转换为毫秒单位输出，直观展示整体耗时。

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