55

最近我开始研究RxJS和 RxJava（来自 Netflix）库，它们致力于响应式编程的概念。

Node.js 在事件循环的基础上工作，它为您提供了异步编程的所有工具，而后续的节点库（如“集群”）可帮助您充分利用多核机器。Node.js 还为您提供了 EventEmitter 功能，您可以在其中订阅事件并对其进行异步操作。

另一方面，如果我理解正确，RxJS（以及一般的反应式编程）的工作原理是事件流，订阅事件流，异步转换事件流数据。

所以，问题是在 Node.js 中使用 Rx 包意味着什么。Node 的事件循环、事件发射器和对 Rx 的流和订阅的订阅有多么不同。

4

2 回答 2

99

Observables 不像 EventEmitters。在某些情况下，它们可能像 EventEmitter 一样，即当它们使用 RxJS Subjects 进行多播时，但通常它们不像 EventEmitters。

简而言之，一个 RxJS Subject就像一个 EventEmitter，但是一个 RxJS Observable是一个更通用的接口。Observables 更类似于零参数的函数。

考虑以下：

function foo() {
  console.log('Hello');
  return 42;
}

var x = foo.call(); // same as foo()
console.log(x);
var y = foo.call(); // same as foo()
console.log(y);

当然，我们都希望看到输出：

"Hello"
42
"Hello"
42

您可以在上面编写相同的行为，但使用 Observables：

var foo = Rx.Observable.create(function (observer) {
  console.log('Hello');
  observer.next(42);
});

foo.subscribe(function (x) {
  console.log(x);
});
foo.subscribe(function (y) {
  console.log(y);
});

输出是一样的：

"Hello"
42
"Hello"
42

这是因为函数和 Observable 都是惰性计算。如果您不调用该函数，console.log('Hello')则不会发生。同样对于 Observables，如果您不“调用”( subscribe)，console.log('Hello')则不会发生。另外，“调用”或“订阅”是一个独立的操作：两个函数调用触发两个单独的副作用，两个 Observable 订阅触发两个单独的副作用。与 EventEmitter 共享副作用并且无论订阅者是否存在都急切执行相反，Observables 没有共享执行并且是惰性的。

到目前为止，函数的行为和 Observable 之间没有区别。这个 StackOverflow 问题可以更好地表述为“RxJS Observables vs functions？”。

有些人声称 Observables 是异步的。那不是真的。如果您使用日志包围函数调用，如下所示：

console.log('before');
console.log(foo.call());
console.log('after');

您显然会看到输出：

"before"
"Hello"
42
"after"

这与 Observables 的行为相同：

console.log('before');
foo.subscribe(function (x) {
  console.log(x);
});
console.log('after');

和输出：

"before"
"Hello"
42
"after"

这证明了订阅foo是完全同步的，就像一个函数一样。

那么 Observable 和函数之间的真正区别是什么？

Observables 可以随着时间的推移“返回”多个值，而函数不能。你不能这样做：

function foo() {
  console.log('Hello');
  return 42;
  return 100; // dead code. will never happen
}

函数只能返回一个值。然而，Observables 可以这样做：

var foo = Rx.Observable.create(function (observer) {
  console.log('Hello');
  observer.next(42);
  observer.next(100); // "return" another value
  observer.next(200);
});

console.log('before');
foo.subscribe(function (x) {
  console.log(x);
});
console.log('after');

带同步输出：

"before"
"Hello"
42
100
200
"after"

但您也可以异步“返回”值：

var foo = Rx.Observable.create(function (observer) {
  console.log('Hello');
  observer.next(42);
  observer.next(100);
  observer.next(200);
  setTimeout(function () {
    observer.next(300);
  }, 1000);
});

带输出：

"before"
"Hello"
42
100
200
"after"
300

总而言之，

  • func.call()意思是“立即（同步）给我一个值”
  • obsv.subscribe()意思是“给我价值。也许很多，也许是同步的，也许是异步的”

这就是 Observables 是函数的泛化（没有参数）。

于 2016-02-06T09:04:19.497 回答
11

侦听器什么时候连接到 Emitter ？

使用事件发射器，只要有他们感兴趣的事件发生，侦听器就会收到通知。当事件发生后添加新的侦听器时，他将不知道过去的事件。新的监听器也不会知道之前发生的事件的历史。当然，我们可以手动编程我们的发射器和监听器来处理这个自定义逻辑。

使用响应式流，订阅者可以获取从一开始就发生的事件流。所以他订阅的时间并不严格。现在他可以对流进行各种操作，得到他感兴趣的事件子流。

这样做的好处就出来了：

  • 当我们需要处理随时间发生的事件时
  • 它们发生的顺序
  • 事件发生的模式（假设每次 Google 股票的买入事件之后，Microsoft 股票的卖出事件都会在 5 分钟内发生）

高阶流：

高阶流是“流的流”：其事件值本身就是流的流。

对于事件发射器，一种方法是将相同的侦听器附加到多个事件发射器。当我们需要关联发生在不同发射器上的事件时，它变得复杂。

使用反应流，这很容易。mostjs的一个例子（它是一个响应式编程库，类似于 RxJS，但性能更高）

const firstClick = most.fromEvent('click', document).take(1);
const mousemovesAfterFirstClick = firstClick.map(() =>
    most.fromEvent('mousemove', document)
        .takeUntil(most.of().delay(5000)))

在上面的示例中，我们将单击事件与鼠标移动事件相关联。当事件作为流可用时，跨事件推断模式变得更容易完成。

话虽如此，使用 EventEmitter，我们可以通过过度设计我们的发射器和侦听器来完成这一切。它需要过度工程，因为它不首先用于此类场景。而反应流之所以如此流畅地做到这一点，是因为它旨在解决此类问题。

于 2016-07-22T12:08:51.877 回答