Effect 的 Stream.throttle 为什么没有逐元素限速？

六个元素放进 Stream.throttle，限速设成每 300 毫秒两个，结果六个元素几乎同时冒了出来，约在 600ms 处，一点限速的效果都没有。

原因是 v4 的 Stream 内部传递的是非空数组块，throttle 的 cost 收到的是整块，不是单个元素。想逐元素限速，先 Stream.rechunk(1)，并且让 cost 返回 chunk.length。

这个行为的根源是分块本身：每个元素走一次拉取协议太贵，所以一次拉一块来摊薄开销。只要搞清楚块长什么样、哪些算子按块工作，限速和成批这两类写法就都不会再出错。下面按你会接着问的顺序展开。

v4 的 Stream 里，块到底是什么？

v3 的 Stream 内部传递 Chunk，v4 里传递的是非空数组。这在 API 上到处能看到：

  • Stream.runCollect 直接返回普通数组。
  • Stream.grouped(n) 的元素是 NonEmptyReadonlyArray。
  • Stream.mapArray 可以一次处理一整块。
  • Stream.DefaultChunkSize 是 4096。

另外，用 Chunk.empty<Event>() 这类 v3 写法的资料，在 4.0.5 上已经不能用了。想改变块大小，用 Stream.rechunk。

为什么 throttle 没有按元素限速？

Stream.throttle 的 cost 函数收到的参数是一整块。下面的写法才是逐元素限速：

import { Effect, Stream } from "effect"

const stamp = (t0: number) => (i: number) =>
  Effect.sync(() => console.log(`${i} @ ${Date.now() - t0}ms`))

export const throttled = Effect.gen(function* () {
  const t0 = Date.now()
  yield* Stream.range(1, 6).pipe(
    Stream.rechunk(1),
    Stream.throttle({
      units: 2,
      cost: (chunk) => chunk.length,
      duration: "300 millis"
    }),
    Stream.runForEach(stamp(t0))
  )
})

不加 rechunk(1) 时，六个元素在同一块里，整块只被延迟一次，所以六个元素同时在约 600ms 处出现。加了 rechunk(1) 后，前两个立即通过，之后大约每 150ms 一个。

所以规则是两条：cost 按 chunk.length 算；想要逐元素限速，就先 rechunk(1)。

为什么 cost: () => 1 是个安静的 bug？

不少资料里写的是 cost: () => 1。在 v4 里它的意思是「每块算一个」，而不是「每个元素算一个」。块里有几个元素它都只扣 1 个单位，限速就悄悄失效了。

想按批处理，用什么算子？

按块计算不总是坑，成批处理恰恰需要它。Stream.grouped(100) 凑够 100 个一批；Stream.groupedWithin(100, "5 seconds") 则是「100 个或 5 秒，先到先发」。后者是批量入库、批量上报的标准件：

import { Effect, Stream } from "effect"

declare const insertMany: (rows: ReadonlyArray<number>) => Effect.Effect<void>

export const batchInsert = (events: Stream.Stream<number>) =>
  events.pipe(
    Stream.groupedWithin(100, "5 seconds"),
    Stream.mapEffect((batch) => insertMany(batch)),
    Stream.runDrain
  )

如果你的数据来自分页接口，想看完整的分页加成批写库写法，见 Effect 里怎么把分页接口变成一条 Stream？。如果你想先弄清 Stream 的背压和适用场景，见 Effect 的 Stream 是什么，背压为什么能让内存有界？。

证据说明：以上行为与数字（约 600ms 同时出现、rechunk(1) 后约每 150ms 一个、Stream.DefaultChunkSize 为 4096）来自对 Effect 4.0.5 的记录，我没有在这次整理中重新运行。groupedWithin 的代码未单独运行，语义取自原文。