ZIO 实战：环境依赖注入、错误通道与并发模型

ZIO[R, E, A] 的含义：
□ R（Environment）：执行它需要哪些依赖（可被 ZLayer 提供）
□ E（Error）：可能失败的类型（可恢复错误走这里）
□ A（Success）：成功值的类型
值构造速查：
□ ZIO.succeed(a)          → 纯值
□ ZIO.attempt(f)          → 可抛异常的计算 → E = Throwable
□ ZIO.fail(e)             → 主动失败
□ ZIO.foreach / foreachPar → 集合迭代
□ for 推导：flatMap/map 的组合即依赖传递

import zio.*
// 类型里写着依赖：需要 Clock，可能失败为 String，成功给 Long
val nowEpoch: ZIO[Clock, String, Long] =
  Clock.currentTime(TimeUnit.MILLISECONDS)
    .mapError(e => s"读时钟失败: ${e.getMessage}")
// 依赖被「暂时缺省」：类型告诉调用者还缺什么
val needsConfig: ZIO[AppConfig, Nothing, Int] =
  ZIO.service[AppConfig].map(_.maxRetries)

错误通道语义：
□ E 是可失败通道：类型可枚举，编译器校验分支处理
□ Throwable 变体：ZIO.attempt 包装，E = Throwable
□ 无错误效果：E = Nothing（不可能失败的类型级保证）
与 Either 互操作：
□ ZIO.fromEither / .either  → 把 E 折叠进 A 侧
□ .either：ZIO[R, E, A] → ZIO[R, Nothing, Either[E, A]]
□ .mapError / .flatMapError：转换错误类型
□ .orElseFail / .orElseSucceed：兜底值
□ 错误折叠：catchAll / catchSome / foldZIO
细分错误类型：
□ 基础设施错误（超时/连接）可提升为领域错误

import zio.*
sealed trait UserError
case object NotFound    extends UserError
case class  DbDown(msg: String) extends UserError
val getUser: ZIO[Any, UserError, String] =
  ZIO.succeed(NotFound)                       // 类型化失败
// 与 Either 互操作：把类型化错误折叠到成功侧
val asEither: ZIO[Any, Nothing, Either[UserError, String]] =
  getUser.either
// 只处理部分错误，其余继续上抛
val fallback: ZIO[Any, UserError, String] =
  getUser.catchSome {
    case NotFound => ZIO.succeed("默认用户")
  }

依赖注入核心：
□ ZEnvironment：运行时「值袋」（type → value 的查找表）
□ ZLayer：依赖的「构造工厂」（依赖 A/B → 产出 C）
  ZLayer[In, E, Out]：需要 In 环境，产出 Out 环境
□ ZIO.service[A]：从 ZEnvironment 按类型取服务（编译期检查）
最简 DI：
□ 服务定义为 trait + 其「ZLayer 工厂」
□ main 里用 .provide(layer) 注入
□ provide 是「填入 R」，类型不匹配编译报错
环境组合：
□ layer1 >*> layer2    ：并行组合（都产出）
□ layer1 >>> layer2     ：串行组合（layer2 依赖 layer1）

import zio.*
trait Logger:
  def log(msg: String): UIO[Unit]
object Logger:
  def log(msg: String): UIO[Unit] = ZIO.serviceWithZIO[Logger](_.log(msg))
  val layer: ULayer[Logger] = ZLayer.succeed(new Logger:
    def log(msg: String): UIO[Unit] =
      Console.printLine(s"[logger] $msg").orDie
  )
// 依赖被类型化：效果需要 Logger 才能跑
val program: ZIO[Logger, Nothing, Unit] =
  Logger.log("hello zio")

fiber 操作：
□ .fork：后台启动，返回 Fiber（不阻塞当前）
□ .join：等待结果（若失败则当前也失败）
□ .interrupt：取消（注入 InterruptedException）
□ .await：等 Outcome（成功/失败/取消三态）
结构化并发：
□ ZIO.foreachPar / collectAllPar / zipPar：
  兄弟并行，任一失败或取消 → 全体取消
□ ZIO.race：多个竞争，首个完成者胜，其余取消
□ ZIO.withInterruptMask / .onInterrupt：取消感知与屏蔽
□ Supervisor：管理生命周期，自动做监控
Fiber 本地数据 / 调度：

import zio.*
object FiberDemo extends ZIOAppDefault:
  def run: ZIO[Any, Nothing, Unit] =
    for
      fiber <- (ZIO.sleep(3.seconds) *> ZIO.debug("完成")).fork
      _     <- ZIO.sleep(1.second)
      _     <- fiber.interrupt                 // 取消后台任务
      out   <- fiber.await                     // Outcome
      _     <- out match
                 case Exit.Success(_)    => ZIO.debug("成功")
                 case Exit.Failure(cause) => ZIO.debug(s"失败/取消: $cause")
    yield ()

组合算子：
□ >>>（串行）：L1 >>> L2 表示 L2 消费 L1 的产出
□ >*>（并行）：L1 >*> L2 表示同时提供两者
□ ++ / +：环境合并（求和类型）
□ ZLayer.provide / provideLayer：注入到效果
□ ZLayer.memoize：构造「缓存单例」（图内共享实例）
设计模式：
□ 每服务一个 module（trait Service + object.layer）
□ 组装层：汇总所有子 layer 成一个 app layer
□ 测试层：测试替换实现，不影响结构

import zio.*
trait DB:
  def query(sql: String): UIO[List[String]]
object DB:
  val layer: ZLayer[Config, Throwable, DB] =
    ZLayer.scoped {
      ZIO.acquireRelease(DBLive.connect)(_.close)
    }
  class DBLive(conn: AnyRef) extends DB:
    def query(sql: String): UIO[List[String]] = ZIO.succeed(List("row"))
object Config:
  val live: ULayer[Config] = ZLayer.succeed(Config("jdbc://..."))
  case class Config(url: String)
// 串行组装：Config → DB → 服务
val appLayer: ZLayer[Any, Throwable, DB] =
  Config.live >>> DB.layer
val program: ZIO[DB, Nothing, Unit] =
  DB.query("SELECT 1").flatMap(rows => ZIO.debug(s"rows=$rows"))

Scope 模型：
□ Scope：一组「延迟注册的 finalizer」的容器
□ acquireRelease(acquire)(release)：获取/释放对
□ 效果运行在一个隐式 Scope 里：成功/失败/取消都触发 finalizer
□ ZIO.scoped：显式打开子 scope（作用域内资源离开即释放）
finalizer 规则：
□ 注册顺序 = 释放逆序（先注册的后释放，像栈）
□ finalizer 永不吞掉主错误（主错误优先传播）
□ 可取消：scope 被取消 → 所有 finalizer 运行
常见资源：
□ 连接池：acquireRelease + memoize
□ 文件/流：ZIO.fromAutoCloseable

import zio.*
object ScopeDemo extends ZIOAppDefault:
  def run: ZIO[Any, Nothing, Unit] =
    ZIO.scoped {
      for
        conn <- acquireRelease // 作用域内有效
        _    <- useConn(conn)  // 离开作用域自动释放
      yield ()
    }
  def acquireRelease: ZIO[Scope, Throwable, String] =
    ZIO.acquireRelease(ZIO.debug("打开连接").as("conn"))(_ =>
      ZIO.debug("关闭连接").orDie
    )
  def useConn(c: String): UIO[Unit] =
    ZIO.debug(s"使用 $c")

运行时组成：
□ Runtime：环境（Environment）+ 调度器（Executor）+ 配置
□ Executor：线程池抽象（类似 Cats Effect 的 compute/blocking）
□ 默认 Runtime：ZIOAppDefault 提供（可覆盖 run）
□ 纤程调度：非阻塞 fiber 挂在协作队列，异步边界让出线程
Executor 分工：
□ 默认池：跑绝大多数 fiber（JVM 线程池，WorkStealing 风格）
□ blocking 池：ZIO.attemptBlocking / blocking 的宿主机
□ 自定义：Runtime.setExecutor / Runtime.setBlockingExecutor
手动运行：
□ Unsafe.unsafe { implicit u => runtime.unsafe.run(program) }
□ 测试：TestRuntime / Live 替换时钟

import zio.*
object RuntimeDemo extends ZIOAppDefault:
  def run: ZIO[Any, Nothing, Unit] =
    // 阻塞调用明确走 blocking 池，不饿死主池
    ZIO.attemptBlocking(Thread.sleep(200))
      .as("阻塞完成")
      .flatMap(ZIO.debug) *>
    ZIO.sleep(10.millis) *> ZIO.debug("异步 fiber 继续")

语义对比：
□ 类型参数：ZIO[R, E, A] 三参 vs IO[A]（Cats 单参，环境走 Reader/依赖注入）
□ 错误通道：ZIO 内建 E（可枚举类型）vs Cats 用 Throwable/Either
□ 依赖注入：ZIO 内建 ZLayer vs Cats 生态用 ReaderT/构造器注入
□ fiber 取消：ZIO interrupt（Exit 三态）vs Cats MonadCancel
□ 资源：ZIO Scope vs Cats Resource
□ 调度：两者都是纤程 + 双线程池，机制接近
工程取舍：
□ ZIO 优点：类型化错误 + ZLayer 让「架构」可编译期验证
  · 测试替身、模块边界非常自然
□ Cats 优点：生态更纯（Cats/cats-effect 被 fs2/http4s/Doobie 广泛使用）
  · 更贴近函数式类型类传统，组合子更「代数」
□ 团队视角：ZIO 上手曲线平（像命令式+类型保护），Cats 更数学

选型速判：
□ 需要类型化错误 + 服务式 DI → 倾向 ZIO
□ 团队函数式基础强 → 两者皆可，Cats 更自然
□ 大型服务式应用 → ZIO 的 ZLayer 模块化收益大

服务分层模式：
□ domain：领域类型（ADT、规则）——纯、零依赖
□ service：业务服务（trait + layer），依赖 domain 与其他 service
□ infrastructure：DB/HTTP/消息 实现（layer，可替换）
□ app 装配：main 组装所有 layer，provide 给入口
测试模式：
□ 测试 layer：内存实现替代 DB/HTTP
□ 时间控制：TestClock 让 sleep/超时可确定性推进
生产就绪：
□ 配置：zio-config 解析环境/文件
□ 日志：ZIO 结构日志（zio-logging，MDC 经 FiberRef）

import zio.*
import zio.test.*
import zio.test.Assertion.*
object ServiceSpec extends ZIOSpecDefault:
  def spec = suite("service")(test("读用户") {
    // 用测试 layer 替换真实 DB
    for
      u <- getUser(1)
    yield assertTrue(u == Some("Alice"))
  }).provide(TestDB.layer)   // 测试替身注入
  def getUser(id: Int): ZIO[DB, Nothing, Option[String]] =
    DB.query(s"SELECT name FROM user WHERE id=$id").map(_.headOption)