forEach 运算符在没有订阅的情况下进行评估

Question

我正在尝试自学一些反应式函数式编程。这个来自 Ben Lesh 的 video 展示了一个 observable 的例子。我之前的阅读表明 observable 是惰性的，即它仅在订阅后才进行评估。奇怪的是，此代码不需要订阅即可打印到控制台。

var Rx = require('rxjs/Rx')

var source = Rx.Observable.from([1,2,3,4,5]);

var newSource = source.filter(x => x % 2 === 1)
                .map(x => x + '!')
                .forEach(x => console.log(x));

来自RxJS docs：

似乎 Observable 必须积极解决 .forEach 发出的承诺，我很对此感到困惑。

进一步的混乱源于这段代码：

var Rx = require('rxjs/Rx')

var source = Rx.Observable.from([1,2,3,4,5]);

var newSource = source.filter(x => x % 2 === 1)
      .map(x => x + '!')
      .do(x => console.log(x));

在运行newSource.subscribe(); 之前不会评估，请帮我解释一下这两个运算符背后的区别。

Answer 1

Observables 默认是惰性的。如果你在一个 observable 上执行一个操作符，在底层，rxjs 会为你创建一个新的 observable 链接到前一个 observable。知道 observables 是不可变的。

但是，ForEach 是一种特殊的运算符。它不会返回一个新的 Observable，但它会在底层订阅 observable，并对该 observable 发出的每个元素执行一个函数。如果您检查 Observable 类本身的 forEach 实现的源代码，您将看到以下内容（只是一个 sn-p）。

const subscription = this.subscribe((value) => {
    if (subscription) {
      // if there is a subscription, then we can surmise
      // the next handling is asynchronous. Any errors thrown
      // need to be rejected explicitly and unsubscribe must be
      // called manually
      try {
        next(value);
      } catch (err) {
        reject(err);
        subscription.unsubscribe();
      }

在这里我们可以看到 observable 正在被订阅并且值是 'next'-ed。下一个函数是您传递给 forEach 调用的函数。