Effect 里怎么把分页接口变成一条 Stream？

分页接口一页接一页，如果先把所有页拼进一个数组再处理，内存就得装下全部数据；手写 while 循环又要自己管游标、出错和取消。

用 Stream.paginate(初始游标, 取一页)：回调返回「本页元素数组」和「下一个游标的 Option」，Option.none() 表示结束；之后接 Stream.groupedWithin 成批、Stream.mapEffect 写库、Stream.runDrain 消费。

这样做的原因是 Stream 是拉取式的：下游处理完一批，上游才去拉下一页，不会无限堆积（Stream 的整体模型见Effect 的 Stream 是什么，背压为什么能让内存有界）。下面是完整代码，以及读它时值得注意的三件事。

完整代码长什么样？

import { Effect, Option, Stream } from "effect"

interface Page {
  readonly items: ReadonlyArray<string>
  readonly next: string | null
}

declare const fetchPage: (cursor: string | null) => Effect.Effect<Page, FetchError>
declare const saveBatch: (rows: ReadonlyArray<string>) => Effect.Effect<void, DbError>

interface FetchError { readonly _tag: "FetchError" }
interface DbError { readonly _tag: "DbError" }

const allItems = Stream.paginate(null as string | null, (cursor) =>
  fetchPage(cursor).pipe(
    Effect.map((page) => [
      page.items,
      page.next === null ? Option.none() : Option.some(page.next)
    ] as const)
  )
)

export const importAll = allItems.pipe(
  Stream.groupedWithin(100, "2 seconds"),
  Stream.mapEffect(saveBatch, { concurrency: 2 }),
  Stream.runDrain
)

fetchPage 和 saveBatch 是占位声明，换成你自己的拉页与写库函数即可。Stream.groupedWithin(n, duration) 按「满 n 个或到时间」成批，适合批量写入。

内存到底占多少？

没有任何地方持有「全部数据」：一页进来、成批写出，内存只与批大小和并发度有关。这正是背压的效果：saveBatch 慢，拉页就跟着慢。

错误会怎么传播？

错误通道是类型化的：importAll 的 E 是 FetchError | DbError，拉页失败和写库失败都会让整条流失败。需要「单条失败不影响其他」时，要在每个元素内部把失败转成值，而不是在最外层兜底。

被取消或超时会怎样？

如果外层把它 Effect.timeout 或中断了，拉页和写库会一起被中断，不需要手动清理循环。

mapEffect 的并发要怎么设？

Stream.mapEffect 不指定 concurrency 时是一个接一个顺序执行；想并发必须显式写，并且要有上限，例如上面的 { concurrency: 2 }。并发度的取舍见并发。

本文代码与说明整理自本站 Effect 教程中关于 Stream 的章节，没有在本地重新运行；其中的类型签名（fetchPage、saveBatch 为占位声明）和行为描述未经逐项验证，具体 effect 版本号原章节未标明，使用前请对照你所用的版本。