Cats Effect 深度解析：IO 运行时、fibers 与并发原语

IO 求值链条：
□ IO.pure(a)      → 纯值，直接返回
□ IO.delay(f)     → 惰性副作用（计算型）
□ IO.blocking(f)  → 阻塞副作用（走 blocking 池）
□ IO.async_[...]  → 外部回调桥接（注册、等待、恢复）
□ 组合子          → flatMap/parMapN/race 等编织成执行图
运行时展开步骤：
1. 解释 AST：把 flatMap 链条编译为「步进器」（步进循环）
2. 分配 fiber：绑定到 WorkStealing 线程池的一个工作线程
3. 步进：每次 flatMap 边界是一个「步」，步间可被抢占/让出
4. 注册回调：异步边界（async/IO.blocking）挂起 fiber 并释放线程

运行时线程模型：
□ compute 池：WorkStealingThreadPool（默认 = CPU 核数）
  - 每个工作线程一个本地队列（LIFO），全局队列 FIFO
  - 本地队列空了才窃取别的线程的尾部任务（降低竞争）
  - 专跑「计算型」fiber，禁止放阻塞调用
□ blocking 池：默认 64 上限，按需增长
  - IO.blocking / IO.interruptible 的宿主机
  - 增长上限可配（blockingExecutionContext）
□ 全局 IORuntime：进程级单例（IORuntime.global）
  - 含 compute/blocking/scheduler(定时器) 三件套
  - 也可为测试/子域建独立 Runtime

// 自定义运行时（测试、共享线程、配额管理）
import cats.effect.*
import cats.effect.unsafe.implicits.global
object RuntimeDemo extends IOApp.Simple:
  def run: IO[Unit] =
    // IO.blocking 自动切换到 blocking 池
    IO.blocking(Thread.sleep(100)).as("done")
      .flatMap(s => IO.println(s"compute 池干的活: $s"))

Fiber 语义：
□ 创建：fiber = IO.start / start 显式、parMapN/race 隐式
□ 取消：fiber.cancel 注入 CancellationException（非堆栈式）
□ 协作点：异步边界（async）、cede、sleep 是「可取消点」
□ 自动取消：par/race 中「兄弟失败 → 整体取消」= 结构化并发
取消传播：
父 fiber 被取消 → 递归取消所有子 fiber
→ 沿运行栈抛 CancellationException
→ 触发每个 bracket/finalizer 的 release
→ 恢复点：可取消操作挂起时立即响应
取消边界（uncancelable）：
□ 需要原子性的临界区必须屏蔽取消（如分配+注册两步）
□ 用 MonadCancel#uncancelable 包住，内部再用
  .onCancel / IO.canceled 感知取消

import cats.effect.*
import cats.effect.kernel.Outcome
import scala.concurrent.duration.*
object FiberDemo extends IOApp.Simple:
  def run: IO[Unit] =
    for
      fiber <- (IO.sleep(5.seconds) *> IO.println("done")).start
      _     <- IO.sleep(1.second)
      _     <- fiber.cancel                     // 5 秒后不会打印
      out   <- fiber.join                       // Outcome.Canceled
      _     <- out match
                 case Outcome.Succeeded(_) => IO.println("完成")
                 case Outcome.Canceled()    => IO.println("被取消")
                 case Outcome.Errored(e)    => IO.raiseError(e)
    yield ()

Ref[A]：原子引用（等价 AtomicReference）
□ get / set / update / modify（modify = getAndUpdate 的纯函数版）
□ 组合：update 在并发下可能重试（CAS 循环），副作用不能放 update 里
□ 适合：计数器、状态快照、缓存位
Deferred[A]：一次性「信号灯」（promise）
□ 一个写者 complete，多个读者 get 挂起等待
□ 无值则 get 挂起（fiber 被 park，不占线程）
□ 适合：屏障、一次性初始化、跨 fiber 通知
Queue[F, A]：有界/无界队列（cats.effect.std.Queue）
□ offer 在无界队永返回；有界队满了挂起 = 天然背压
□ take 空队挂起；tryOffer/tryTake 非阻塞探测
□ 还有 bounded 有界、dropping/sliding 策略变体

import cats.effect.*
import cats.effect.std.{Queue, Deferred}
import cats.syntax.all.*
object PrimitiveDemo extends IOApp.Simple:
  def run: IO[Unit] =
    for
      q     <- Queue.bounded[IO, Int](10)
      done  <- Deferred[IO, Unit]
      _     <- (produce(q, done)).start
      _     <- (consume(q, done)).start
      _     <- done.get           // 等生产消费完成
    yield ()
  def produce(q: Queue[IO, Int], d: Deferred[IO, Unit]): IO[Unit] =
    (1 to 5).toList.traverse_(q.offer) *> d.complete(()).void
  def consume(q: Queue[IO, Int], d: Deferred[IO, Unit]): IO[Unit] =
    d.get *> (1 to 5).toList.traverse_(_ => q.take.flatMap(IO.println))

四种背压/限流策略：
□ 挂起（Blocking）：有界 Queue 满则 offer 挂起
  → 天然最稳：生产者被拖慢到消费速率，不丢数据
□ 丢弃（Dropping）：满则丢新（或丢旧 Sliding）
  → 有损但零等待，适合指标/日志这类「可丢」数据
□ 合并/采样：Semaphore 限并发，合并小请求
  → 控制 in-flight 数量而非速率
□ 令牌桶：自实现 RateLimiter，按固定速率放行
工程取舍：
□ 数据库写：挂起 + 重试（不能丢）
□ 实时指标：dropping/sliding（丢一点无所谓）
□ 外部 API：Semaphore 限制并发 + 退避重试
□ 消息队列：Queue + 消费者 fiber 组（worker pool）

import cats.effect.*
import cats.effect.std.{Queue, Semaphore}
import cats.syntax.all.*
object BackpressureDemo extends IOApp.Simple:
  def run: IO[Unit] =
    for
      sem <- Semaphore[IO](4)                 // 最多 4 个 in-flight
      _   <- List.fill(20)(job(sem)).parTraverse_(identity)
    yield ()
  def job(sem: Semaphore[IO]): IO[Unit] =
    sem.permit.use(_ => IO.blocking(Thread.sleep(50)))
      .handleErrorWith(t => IO.println(s"fail: ${t.getMessage}"))

bracket 结构：
acquire.flatMap(a => use(a).guarantee(release(a)))
□ acquire：获取资源（可取消）
□ use：使用资源（可被取消/出错）
□ release：无论成功/失败/取消都执行（不可取消的清理）
Resource[F, A]：bracket 的「可组合」形态
□ acquire / release 分层注册
□ 组合：两个 Resource 自动嵌套释放（先 inner 后 outer）
□ 与 FlatMap 组合：for 内 yield 多个资源
MonadCancel 提供的取消感知：
□ uncancelable：临界区屏蔽取消
□ onCancel / guaranteeCase：感知取消做针对性清理
□ IO.canceled：在屏蔽区里读取「是否收到取消请求」

import cats.effect.*
import cats.effect.std.Console
object ResourceDemo extends IOApp.Simple:
  def run: IO[Unit] =
    (conn.pooled.use { pool =>
      pool.query("SELECT 1").flatMap(rows => IO.println(s"rows: $rows"))
    }).as(())
  // 分层资源：连接池 + 单连接
  val conn: Resource[IO, ConnPool] =
    Resource.make(IO.println("创建池"))(_ => IO.println("关闭池"))
      .flatMap(pool => pool.connection.map(pool.withConn))
  final case class ConnPool(withConn: Conn)
  final case class Conn(query: String => IO[Int])

超时实现：
□ IO.timeout(d)：内部 = race(fa, sleep(d)) 
  → 超时侧赢则取消 fa，抛 TimeoutException
□ 取消时机：只有「可取消点」能中断；纯计算超时不灵
□ 超时后的清理：fa 里的 bracket 仍会释放资源
重试模式：
□ retryN / retryA：指数退避 + 抖动（jitter）
□ 只对「可恢复错误」重试：先 map 成 Either，过滤
□ 别对 CancellationException 重试（那是取消不是失败）
□ 幂等性：重试前提是操作可重复执行
熔断与并发控制：
□ Semaphore + timeout 组合 = 慢调用隔离
□ 滑动窗口错误率 → 打开熔断 → 快速失败（半开试探）

import cats.effect.*
import scala.concurrent.duration.*
object TimeoutDemo extends IOApp.Simple:
  def run: IO[Unit] =
    val slow = IO.blocking(Thread.sleep(3000))
    slow.timeoutTo(1.second, IO.println("慢调用被熔断，返回兜底"))
      .recoverWith { case e: java.util.concurrent.TimeoutException =>
        IO.println(s"捕获超时: ${e.getMessage}")
      }.void

调度控制原语：
□ IO.async / async_：把外部回调接入 fiber 世界
  → 回调必须在「某个线程」恢复 fiber（通常用 compute 池）
  → 这是唯一能跨线程「唤醒挂起 fiber」的机制
□ IO.cede：让出当前步，回到队列尾部
  → 协作式公平：给同优先级 fiber 运行机会
  → parMapN 内部自动插 cede，避免一个 fiber 独占
□ IO.cede 在循环里 = 手动 yield
□ 抢占：cancel 只能中断「挂起点」；纯 CPU 循环不可中断
async 的正确姿势：
□ 注册回调一次（幂等）
□ 回调返回 Either[Throwable, A]
□ 用 cb.resume / cb.raiseError（CE3 提供的回调包装）
□ 别在回调里做重活：放 compute 池

import cats.effect.*
object AsyncDemo extends IOApp.Simple:
  def run: IO[Unit] =
    externalCall.flatMap(x => IO.println(s"回调结果: $x")).void
  // 桥接一个异步回调 API
  def externalCall: IO[String] =
    IO.async_ { cb =>
      legacyCallback { value =>
        cb(Right(value))     // 在 compute 池上恢复 fiber
      }
    }
  def legacyCallback(f: Either[Throwable, String] => Unit): Unit =
    new Thread(() => f(Right("legacy-ok"))).start()

性能调优清单：
□ 减少 async 边界：flatMap 链比每步 async 快一个量级
□ 避免阻塞 compute：blocking 走独立池（防线程池耗尽）
□ 控制 fiber 数量：Semaphore 限并发，防「线程池被几千 fiber 挤爆」
□ 批量 IO：DB 用 batch、网络用 pipelining，减少 round-trip
□ 缓冲：Queue 加 batch 消费者（一批 take 再处理）
□ 内存：避免大对象在 fiber 栈间搬移（WorkStealing 的窃取有缓存代价）
机制深入：
□ 每步开销：WorkStealing 步进器把 flatMap 内联优化成循环
□ 监视：IO.println / 自建计数器看 fiber 挂起分布
□ 配置：IORuntime.global 可替换 compute/blocking 线程数
□ 工具：sbt-jmh 做微基准，别用 println 计时

性能心智模型：
并发度 = 资源受限值（DB 连接数、外部 API 配额）
吞吐  = 并发度 × 单请求速率
延迟  = 排队时间 + 处理时间（并发过高反而变慢）