no-stream 似乎比 js 原生数组方法快

no-stream

no-stream 又是一个处理集合的库，介绍它不如直接拿来和原生数组比一比。

用 benchmark 测试吧

以下是部分测试代码，它测试了，长度为100 ～ 100,000的数组经历2 ～ 5次的 map 后再 reduce ，这些情况下的 ops/sec （每秒完成次数）。

new Suite()
    .add("array", function () {
      let d = data;
      for (let i = 0; i !== map_count; i++) {
        d = d.map(mf);
      }
      d.reduce(rf, 0);
    })
    .add("no-stream", function () {
      let d = ns(data);
      for (let i = 0; i !== map_count; i++) {
        d = d.map(mf);
      }
      d.reduce(rf, 0);
    })

测试结果

array

no-stream

比原生数组方法更快！

可以看出，除了数据量在 100 且 map 2次时 no-stream 比较慢，其他情况都是 no-stream 更快。并且随着数据规模和转换次数的增长，no-stream 会比 array 快更多！

在本地进行测试

git clone https://github.com/Iplaylf2/no-stream.git
cd no-stream
npm run init
npm run init-debug
npm run benchmark

没有魔法

为什么 no-stream 会更快？这其中没有用到什么黑科技，也没有魔法般的技艺，仅仅是 no-stream 只遍历了1次数组。

只遍历1次数组

以下有两段代码，s 是 Array 的一个实例。

a

s.map(aa).map(bb).forEach(cc);

b

for (var x of s) {
  x = aa(x);
  x = bb(x);
  cc(x);
}

他们做的事情是一样的，但是 b 版本会更快。原因如下。

• a 总共遍历3次，b 只遍历了1次。每次遍历的子过程，都会有边界判断。
• a 会产生中间数组，会使用更多的内存空间。

没有魔法

no-stream 很普通，它还没学会怎么把链式调用的代码转化为一个循环，用来解决所有数据遍历的问题。

它底层的底层用到的是 transduce 的变体。

transduce 想法是把集合的一系列转换（transform）方法，预处理压缩为1个转换方法，然后在消费时（reduce）只遍历1次数据只做1次消费。

把层层遍历，变成1次遍历中的层层转换。就好像是中间件。

transduce in js

transduce 简化下来的核心用代码表达是这样的。

/**
 * conj 用来合并不同的转换函数
 * @param tf1 转换函数1
 * @param tf2 转换函数2
 * @returns 新的转换函数
 */
function conj(tf1, tf2) {
  return (next) => tf1(tf2(next));
}

/**
 * reduce 采用最终的转换函数，并且消费
 * @param source 数据源
 * @param tf 转换函数
 * @param rf 消费函数
 */
function reduce(source, tf, rf) {
  const transduce = tf(rf);
  for (const x of source) {
    const continue_ = transduce(x);
    if (!continue_) {
      break;
    }
  }
  return rf.result;
}

// 演示

s.map(aa).map(bb).forEach(cc);

// 就相当于

const tf = conj(map(aa), map(bb));
reduce(s, tf, forEach(cc));

tf 真的很像中间件呢。

总之，no-stream 有理由，也在事实上比原生数组方法快。

抽象的 no-stream

no-stream 高效的同时，还有着和流一样的抽象能力。

当我用到流这种数据结构时，会希望它：

• 能通过转换它的元素得到一个新的流。
• 惰性求值，只会求值消费时用到的元素。
• 数据源不是固定的，能在消费时才获取数据，能表达无限长的数据。

no-stream 也有这样的能力。使用 ns 创建流：

import { ns } from "no-stream";

const s = ns(function* () {
  let x = 0;
  while (true) {
    yield x++;
  }
});

const result = s
  .map((x) => x * 2)
  .filter((x) => x % 4 === 0)
  .take(10)
  .reduce((r, x) => r + x, 0);

console.log(result); // 180

codesandbox

为什么叫 no-stream ?

在 js 可以通过 生成器（generator） 方便地构造流。只是把一个 generator 转换成另一个 generator 后，每次迭代都会有额外的检查，在性能上会有所损耗。

而 no-stream 就避免了这个损耗，不使用 generator 一层包一层的结构。这是有代价的，同步的 no-stream 无法控制单个元素的 生成（yield） ，它的消费总是会彻底迭代一个流。

这不过是微弱的代价，反映在 api 上是缺乏 ns.zip 这个函数的实现。

其实更应该叫 no-generator 吧。

终于有 lazy 的 groupBy 了

groupBy 是我觉得最有趣的方法了，在去年我就在想如何实现一个 lazy 的 groupBy，有了 transduce 的意识终于让我实现成功了。

在对一批数据进行分组后，可以对分组的数据进行“流式”处理吗？如果分组的数据提前消费完，能不能提前对这部分数据进行退出？

接下来看一个 groupBy 的实际使用例子吧。使用 nsr 消费分组后的数据：

import { ns, nsr } from "no-stream";

//  构造一个 1 < x < 10 的随机数流
const random_s = ns(function* () {
  while (true) yield;
})
  .map(() => Math.random())
  .map((x) => 1 + x * 9);

// 对元素向下取整，按该值进行分组，每一组都是以 n 作为开头的随机数，
const result = random_s
  .groupBy(
    (x) => Math.floor(x),
    // 以 n 作为开头
    (n) =>
      nsr<number>()
        // 保留 n 位小数
        .map((x) => x.toFixed(n))
        // 取前 n 个随机数
        .take(n)
        // 把数据作为数组返回
        .toArray()
  )
  // 取前3组
  .take(3)
  // 把数据作为数组返回
  .toArray();

console.log(result);

// 可能的结果

// [ [ '1.7' ], [ '3.487', '3.285', '3.362' ], [ '2.19', '2.91' ] ]

codesandbox

异步的 no-stream

no-stream 也有一套异步版本的 api，使流在 map, filter, take... 过程中也能 await，配合 AsyncGeneratorFunction 食用味道更佳。

使用 ans 创建异步流：

import { ans } from "no-stream";

function delay(span: number) {
  return new Promise((r) => setTimeout(r, span));
}

const s = ans(async function* () {
  while (true) yield;
});

s.map(async () => {
  const x = Math.random();
  await delay(x);
  return x;
})
  .take(3)
  .foreach(async (x) => {
    await delay(10);
    console.log(x);
  });

codesandbox

observable

知道 rxjs 的人对 observable 应该不会陌生，no-stream 也有对 observable 的实现呢，transduce 本身就有一丢 push 的味道在其中。

曾经的我会以为需要额外实现一个 push 的流去表达 observable，实际上有异步流就够了。

使用 ans.ob 创建 observable:

import { ans } from "no-stream";

const s = ans.ob<void>((subscribe) => {
  function listener() {
    subscribe.next();
    // subscribe.complete();
    // subscribe.error(xxx);
  }

  document.body.addEventListener("mousemove", listener); // 订阅鼠标移动事件

  // 返回取消订阅的方法
  return () => document.body.removeEventListener("mousemove", listener);
});

顺便搭上简化版的节流（throttle）和防抖（debounce），做一个小页面吧。演示地址 & 代码地址

尾声

对 no-stream 的介绍已经到了尾声了，感谢大家的阅读。这里是 no-stream 的仓库地址 github.com/Iplaylf2/no… 。

了解更多后，大家会发现 no-stream 的 api 并不多，与 lodash 和 rxjs 相比简直贫乏，连上文说到的 throttle 和 debounce 都没有。

这不是我还没写完，只是我觉得 less is more 。就是因为懒。

each-once

如果要阅读具体的实现原理，可以查看 no-stream 依赖的库 each-once ，地址奉上 github.com/Iplaylf2/ea…。

each-once 更为基础 ，还支持 tree sharking ，如果有人拿来用的话。