筛选、映射和聚合看起来是数组能力，Ruby 也能让自定义任务源获得这些方法。关键是 each 协议。Enumerator 提供另一种控制遍历的方式，Lazy 又把求值推迟到消费时；是否需要全部输入，必须通过需求与调用次数观察。

文章卡

项目 内容
先修 第 04、06–07 篇；集合、yield 与退出目标
实验 bundle exec ruby labs/08/run.rb
核心问题 each 如何支持组合，惰性何时求值，重复消费是否缓存？
验收 自定义集合、100 对 3 的调用计数、重复 force 与有限 take
工程边界 输入可重放性、资源寿命和输出上限必须另行声明

each 是协议的起点

本篇定义一个最小任务流：

1
2
3
4
5
6
7
8
9
10
11
12
13
class TaskStream
  include Enumerable

  def initialize(items)
    @items = items
  end

  def each
    return enum_for(__method__) unless block_given?
    @items.each { |item| yield item }
    self
  end
end

提供 each 并包含 Enumerable 后，可以使用 select、map 和 reduce。对象没有继承 Array，也没有复制 Array 的存储接口；它承诺的是逐项交付。本篇把输入存为数组以保持实验简单，第 14 篇会对照多个实现相同协议的任务源。Enumerable

无 block 时返回 Enumerator，调用者可以选择稍后枚举。enum_for(__method__) 描述如何重新调用当前方法，并不立即复制全部元素。这个惯例使协议可以组合，但是否能够重复读取，仍由底层源的能力决定。

对 [1, 2, 3, 4, 5] 筛选奇数后乘十，结果是 [10, 30, 50]；累加得到 15。lab 对两种组合都断言。方法组合的正确性来自输入、次序、block 返回值和错误传播，不只来自加入一个模块。

内部迭代与外部拉取

source.each { ... } 是方法控制逐项调用 block。enum = source.each; enum.next 则由调用者拉取一项。lab 连续获得 1 与 2，调用 rewind 后再得到 1。Enumerator

这里 rewind 成功是因为这个实验源可重放。若源是 Socket、一次性 IO 或带外部副作用的游标，不能从这个结果推导所有 Enumerator 都能恢复原始数据。枚举器的控制状态与数据源的真实状态是两个层次。

外部拉取结束会产生 StopIteration。调用者可以显式处理结束，也可以使用合适的内部迭代方法；不要用捕获全部 StandardError 的方式判断“数据已读完”，否则解析错误和 I/O 错误也会伪装成正常结束。

立即 map 会先处理全部输入

实验先写一个普通管道：

1
2
3
4
5
6
7
8
eager_count = 0
eager = (1..100).map do |n|
  eager_count += 1
  n * 2
end.take(3)

raise unless eager == [2, 4, 6]
raise unless eager_count == 100

take 在 map 的结果上工作，map 已经遍历一百项并构造结果数组。最终只保留三项，不表示前面的计算只发生三次。对于读取大型输入、昂贵转换或有副作用的 block，这个顺序非常重要。

输出结果相同并不能证明计算成本相同。这里用确定性的调用计数检验遍历数量，比一次很短的墙钟时间更适合说明语义；真正性能比较在第 36、37 篇记录环境、预热和多次样本。

Lazy 将需求传到上游

对应惰性版本：

1
2
3
4
5
6
7
8
9
10
lazy_count = 0
pipeline = (1..100).lazy.map do |n|
  lazy_count += 1
  n * 2
end.take(3)

raise unless lazy_count.zero?
lazy = pipeline.force
raise unless lazy == [2, 4, 6]
raise unless lazy_count == 3

构造管道时没有调用转换 block，force 消费结果时才执行；take 满足三项需求后停止继续读取。在这个没有过滤的实验中，转换次数正好三次。Enumerator::Lazy

若加入 select，得到三项结果可能需要检查更多源元素。对整数源选奇数再 take 四项，至少要读到 7。把“take(3) 就只读三项”作为所有 Lazy 管道的规则是不正确的；需求数与输入读取数取决于各步的语义。

惰性管道也可能永远无法满足需求，例如对无界正整数筛选负数后 take 一项。结果上限不是运行时间上限。输入边界、取消和资源预算仍要由应用设计，第 33 篇会讨论跨进程超时和清理。

重复 force 不保证缓存

lab 对同一个 pipeline 再执行一次 force，计数从 3 增加到 6。这里重新枚举了可重放的 Range，并未自动复用上次结果。

如果转换 block 写日志、更新数据库或修改外部集合，第二次消费可能再次产生这些动作。想要只处理一次，应在调用边界控制消费次数，或者明确物化并保存结果；不要把 Lazy 对象本身当成结果缓存。

也不能反过来认为所有源都会重跑相同内容。一次性文件源可能已经读到末尾，实时输入可能产生新内容，随机源可能返回不同值。写一个可重用接口时，应声明结果是快照、可重放视图还是一次性流。

无界输入只能有限消费

实验安全地组合一个无末端的整数范围：

1
2
result = (1..).lazy.select(&:odd?).take(4).force
raise unless result == [1, 3, 5, 7]

select 与 take 都属于惰性链，有限需求能够终止。若移除 take 再 force，就要求构造无限结果，不能得到正常的有限返回。不要实际运行这种无界物化来“看看会不会结束”；语义和有限反例已经足够说明边界。

lazy 也不会让已完成的 map 变成惰性。写成 (1..100).map { ... }.lazy.take(3)，前面的立即 map 已经发生。对 IO 亦然：先 read 全部文件，再包装成 Lazy，并未减少读取内存。

资源寿命与惰性消费要匹配

若在 File.open block 中构造返回一个依赖该文件的 Lazy 管道，block 结束时文件可能已关闭，调用者再 force 就失败。需要让消费发生在资源有效期内，或建立拥有关闭职责的流对象，并明确提前终止如何清理。第 19 篇将实际验证异常路径和关闭状态。

Taskbook 的早期核心使用已校验的内存 Array，不为三项合成数据引入文件流生命周期。逐步换为批量处理时，再用结果一致性、读取次数和资源清理共同验收；“方法链更长”不是优化证明。

实验与练习

本篇输出 eager=100 lazy_first=3 lazy_second=6、结果 [2, 4, 6] 和有限无界输入的 [1, 2, 3]，最后 PASS 08。它证明指定管道的需求传播与重复枚举行为，不承诺所有数据源的缓存、可重放性或性能。

练习是给 Lazy 管道增加偶数筛选，记录获得前三项结果需要检查几项源数据，再把 take 放在筛选之前比较结果。另一个练习是实现一个只能消费一次的 each 源，分别执行两次 force，给第二次行为定下明确契约和断言。两项练习都关注实际迭代次数，而不只比较最终数组。

参考资料

系列导航

导读 · 上一篇：07：闭包与非局部控制流 · 下一篇：09：作用域、self 与常量：名字怎样绑定到对象 · 完整源码包