JavaScript Iterator Helpers 实战:用惰性迭代器重构大数据处理管道

前阵子帮人看一个日志分析脚本,需求简单到不能再简单:读 Nginx 的 access log,筛出所有 5xx 请求,按耗时从高到低取前十条。

代码大概是这么写的:

const lines = fs.readFileSync('access.log', 'utf8').split('n');

const slow = lines
  .map(parseLine)
  .filter(line => line.status >= 500)
  .map(line => ({ url: line.url, ms: line.ms }))
  .sort((a, b) => b.ms - a.ms)
  .slice(0, 10);

测试文件只有几百行,跑得飞快。换成一份 80MB 的真实日志,进程内存直接冲到 1.6G,Node 还没来得及 OOM,机器先开始疯狂 swap。

问题不在业务逻辑,而在这条链子的形态:map 生成一个和输入等长的数组,filter 再生成一个,第二个 map 又生成一个,sort 还要复制一份出来排序。数据被完整地物化了三到四次,而实际上我们只想要其中十条。

这种写法在数据量小的时候完全没问题,也无所谓。但只要数据规模上去,或者管道前面看一眼就知道后面会丢掉 99% 的元素,中间数组就纯属浪费。

JavaScript 现在给了我们另一种选择:Iterator Helpers(迭代器辅助方法)。

它到底是什么

一句话概括:把以前只长在数组身上的 map、filter、flatMap、reduce 这一套,搬到了迭代器协议上,而且整个过程是惰性的。

感受一下最核心的差别:

const it = [1, 2, 3].values().map(x => {
  console.log('计算了', x);
  return x * 2;
});

// 到这里,控制台一片空白,什么都没发生
console.log('只是拿到了一个迭代器');

// 直到调用终结方法,管道才真正开动
console.log(it.toArray());
// 依次打印:计算了 1 / 计算了 2 / 计算了 3
// 然后得到 [2, 4, 6]

对比一下数组版本:

[1, 2, 3].map(x => {
  console.log('计算了', x);
  return x * 2;
});
// 立刻打印三行,并且立刻返回一个新数组

差别看起来只是执行时机,但实际影响要深得多:

  • 数组方法:立即求值,每一步都产生一个完整的新数组,长度等于当前输入长度。
  • 迭代器辅助方法:惰性求值,每一步返回的还是一个迭代器,不占额外内存;只有调用终结方法时才一个一个地往下取元素。

惰性带来的直接后果就是:管道可以处理无限序列,也可以在满足条件时立刻停下来,后面的元素根本不会被读取。

可用的方法一览

目前标准化并落地的这一批方法,大致分两类。

返回新迭代器的(惰性中间操作):

  • map(fn):逐元素转换
  • filter(fn):逐元素筛选
  • flatMap(fn):转换并展平一层
  • take(n):最多取前 n 个,取满立即终止上游
  • drop(n):跳过前 n 个,剩下的继续流下去

消耗迭代器、返回具体结果的(终结操作):

  • toArray():收集成数组
  • reduce(fn, init):归约成一个值
  • forEach(fn):逐个消费,返回 undefined
  • some(fn) / every(fn) / find(fn):短路查询

另外还有一个静态方法 Iterator.from(),负责把任意可迭代对象或类数组对象包装成一个合规的迭代器。它顺便解决了一个老问题——那些只有 Symbol.iterator 却没继承到辅助方法的第三方迭代器,包一层就能用了:

// 类数组对象也能直接包
const wrapped = Iterator.from({ 0: 'a', 1: 'b', 2: 'c', length: 3 });
console.log(wrapped.toArray()); // ['a', 'b', 'c']

需要说明的是,早期的提案里还有 takeWhile 和 dropWhile,后来在标准化过程中被拿掉了。这两个用生成器自己写也就五六行,后面会给一个例子。

一个必须记住的前提:它不在数组上

这是最容易踩的坑,没有之一。

const result = [1, 2, 3].map(x => x * 2);
// 仍然是数组,用的是 Array.prototype.map

Array.prototype.map 把迭代器版本挡住了。想用迭代器那一套,必须先拿到迭代器:

[1, 2, 3].values().map(x => x * 2).toArray();

常见的取法有这几种:

arr.values()            // 值迭代器
arr.keys()              // 索引迭代器
arr.entries()           // [index, value] 迭代器
someMap.entries()       // Map 的键值对
someSet.values()        // Set 的值
generateSomething()     // 生成器函数返回的就是迭代器

反过来也要注意:迭代器的 map 返回的不是数组。它没有 length,不能下标访问,console.log 出来是一坨 Object [Iterator]。想要数组,老老实实调 toArray()。

把开头那个脚本重写一遍

先准备一个解析函数,日志格式假设是 时间 方法 路径 状态码 耗时ms:

function parseLine(line) {
  const parts = line.split(' ');
  if (parts.length !== 5) return null;
  const [time, method, url, status, ms] = parts;
  const code = Number(status);
  if (Number.isNaN(code)) return null;
  return { time, method, url, status: code, ms: Number(ms) };
}

版本一:整条管道走惰性,只在最后物化一次。

const slow = Iterator.from(lines)
  .map(parseLine)
  .filter(line => line !== null)
  .filter(line => line.status >= 500)
  .map(line => ({ url: line.url, ms: line.ms }))
  .toArray()
  .sort((a, b) => b.ms - a.ms)
  .slice(0, 10);

注意最后三步是普通数组操作。排序天然需要一个完整的集合,这是没办法回避的,所以让它在管道末端物化一次是合理的。关键在于前面那段:整个筛选过程零中间数组,内存里始终只有一个元素在流动。

版本二:如果只是想知道「有没有 5xx」,连排序都不用做。

const hasServerError = Iterator.from(lines)
  .map(parseLine)
  .filter(Boolean)
  .some(line => line.status >= 500);

some 一旦命中就立即返回 true,同时关闭整条管道,后面的行再也不会被解析。对于一天的日志来说,如果错误发生在开头,实际只读了前几十行就结束了。

真正让性价比起飞的是 take

前面说了,take(n) 取满 n 个就终止上游。这一点在「只要几条样本」的场景下威力巨大:

// 找出前 5 条 5xx,找到就停,后面的日志完全不会被解析
const samples = Iterator.from(lines)
  .map(parseLine)
  .filter(line => line !== null && line.status >= 500)
  .take(5)
  .toArray();

放在数组版本里,这个需求几乎无解——你没法让 filter 在中途停下来。除非手写 for 循环加 break,但那样可读性又回去了。

drop 则是 take 的镜像,适合做分页:

function page(items, pageIndex, pageSize) {
  return items
    .values()
    .drop(pageIndex * pageSize)
    .take(pageSize)
    .toArray();
}

这里的 items 甚至可以是一个任意大的可迭代对象,函数本身不会因为数据量而变慢——它只推进到需要的那个位置为止。

和生成器配合,处理无限序列

迭代器辅助方法最舒服的用法,是接在生成器后面。因为生成器本身就是迭代器,天生支持无限序列。

比如找出斐波那契数列里前 6 个能被 3 整除的数:

function* fibonacci() {
  let a = 1;
  let b = 1;
  while (true) {
    yield a;
    [a, b] = [b, a + b];
  }
}

const result = fibonacci()
  .filter(n => n % 3 === 0)
  .take(6)
  .toArray();

console.log(result); // [3, 21, 144, 987, 6765, 46368]

用数组做这件事是不可能的——第一个 map 就会永远跑下去。

再补一个 takeWhile 的手写版本,用生成器委托实现:

function* takeWhile(iterable, predicate) {
  for (const value of iterable) {
    if (!predicate(value)) return;
    yield value;
  }
}

它和其他迭代器辅助方法可以无缝拼接:

const small = takeWhile(fibonacci(), n => n < 1000).toArray();
console.log(small); // [1, 1, 2, 3, 5, 8, 13, 21, 34, 55, 89, 144, 233, 377, 610, 987]

提前退出会触发 return()

这一点知道的人不多,但很有用。

当 take、some、find 这类操作让管道提前结束时,标准要求调用上游迭代器的 return() 方法。对生成器来说,这意味着函数会在 yield 处被终结,finally 代码块会执行。

function* readChunks() {
  try {
    let i = 0;
    while (true) {
      console.log('读取第', i, '块');
      yield `chunk-${i}`;
      i++;
    }
  } finally {
    console.log('资源已释放');
  }
}

const firstTwo = readChunks().take(2).toArray();
console.log(firstTwo);

输出是:

读取第 0 块
读取第 1 块
资源已释放
[ 'chunk-0', 'chunk-1' ]

日志只打了两次,第三次 next() 直接返回 done,紧接着 finally 被执行。这意味着你可以把文件句柄、数据库游标、网络连接的关闭逻辑写在 finally 里,而不必担心有人提前 break 导致泄漏。这一点比手写循环要安全得多。

几个真实的坑

第一,迭代器是一次性的。

const it = [1, 2, 3].values();
console.log(it.next());      // { value: 1, done: false }
console.log(it.toArray());   // [2, 3] —— 不是 [1, 2, 3]

第一条元素已经被 next() 消费掉了。想遍历两次,就得重新创建迭代器。这和数组完全不同,从数组思维切过来的时候特别容易翻车。

第二,别忘记终结方法。

Iterator.from(lines).map(parseLine).filter(Boolean);
// 什么都不会发生,因为没有人去 next()

惰性管道的执行是靠终结操作驱动的。toArray、forEach、reduce、some、find,至少得有一个,而且要能真正跑到结束(some 提前返回 true 的情况除外)。

第三,惰性不等于更快。

这是最需要冷静看待的一点。迭代器辅助方法本质上还是对每个元素做一次函数调用,而 V8 对数组的 map 和 filter 有大量内联和优化。在数据量小(几百几千个元素)、或者你本来就需要一个完整数组的情况下,数组方法很可能反而更快,代码也更直白。

我自己的判断标准大概是这样的:

  • 数据量小,或者明确需要一个数组结果 → 直接用数组方法,别折腾。
  • 数据量大,且只需要其中一小部分 → 迭代器辅助方法,配合 take / some / find。
  • 上游是生成器、流、或者无法一次性放进内存的数据 → 只能用迭代器。
  • 需要多次遍历同一份数据 → 先 toArray() 落成数组再玩,重复消费迭代器只会得到空结果。

第四,异步迭代器还没跟上。

迭代器辅助方法目前是同步的。如果你的数据源是 async generator 或者流式读取,那一套辅助方法还不能直接用,需要写成 for await...of 的形式手动处理,或者用社区库。异步版本还在提案阶段,暂时别指望。

兼容性和降级

这个特性目前已经在主流浏览器的较新版本中默认可用,Node 端也跟上了。但考虑到用户环境的碎片化,上线前最好还是做一次特性检测:

function hasIteratorHelpers() {
  if (typeof Iterator === 'undefined') return false;
  if (typeof Iterator.prototype.map !== 'function') return false;
  if (typeof Iterator.prototype.take !== 'function') return false;
  return true;
}

如果检测不通过,有两条路。

一是引入 polyfill。core-js 里已经包含了完整的实现,直接按需引入即可,改动成本几乎为零。

二是自己写一层很薄的封装,用生成器兜底。对于只在项目内部使用的场景,这样反而更轻:

function* mapIter(iterable, fn) {
  for (const value of iterable) yield fn(value);
}

function* filterIter(iterable, fn) {
  for (const value of iterable) {
    if (fn(value)) yield value;
  }
}

function* takeIter(iterable, n) {
  if (n <= 0) return;
  let count = 0;
  for (const value of iterable) {
    yield value;
    count++;
    if (count >= n) return;
  }
}

用的时候把调用顺序换个方向就行,惰性特性完全保留:

const samples = Array.from(
  takeIter(filterIter(mapIter(lines, parseLine), line => line && line.status >= 500), 5)
);

可读性确实差一些,但作为过渡手段足够了。

回到那个 80MB 的脚本

改完之后,脚本的峰值内存从 1.6G 降到了不到 200M,主要开销变成了 readFileSync 读进来的那整份字符串本身。

如果连这一步都想去掉,可以再往前推一层,用 readline 或者流式读取逐行产出,然后接同一条迭代器管道。这样整个流程就是真正的流式处理了,无论日志多大,内存占用都保持恒定。

回头想想,这个改动的价值其实不在于省了多少内存,而在于它逼着我重新想清楚一件事:这条管道里,每一步是必须完整处理所有数据,还是只需要看到一小部分就能得出结论?

数组方法的默认答案永远是「完整处理」,而迭代器辅助方法给了你说「不一定」的机会。对于数据分析、日志处理、搜索匹配这些「找到就够了」的场景,这个差别可能就是能不能跑起来的分界线。

JavaScript Iterator Helpers 实战:用惰性迭代器重构大数据处理管道
收藏 (0) 打赏

感谢您的支持,我会继续努力的!

打开微信/支付宝扫一扫,即可进行扫码打赏哦,分享从这里开始,精彩与您同在
点赞 (0)

版权声明:
本站资源有的来自互联网收集整理,本站纯免费分享提供学习使用,如果侵犯了您的合法权益,请发送邮件1506151422@qq.com联系,将会及时下架删除。
本站资源仅供研究、学习交流之用,免费开源项目不代表完全可商用,若商业用途请先咨询开发企业能否商用,否则产生的一切后果将由下载用户自行承担。
原创板块未经允许不得转载,否则将追究法律责任。

淘吗网 javascript JavaScript Iterator Helpers 实战:用惰性迭代器重构大数据处理管道 https://www.taomawang.com/web/javascript/2857.html

常见问题

相关文章

猜你喜欢
发表评论
暂无评论
官方客服团队

为您解决烦忧 - 24小时在线 专业服务