javascript 响应式编程 - Node.js 中的 RxJS 与 EventEmitter

Reactive Programming - RxJS vs EventEmitter in Node.js

提问by Kishore Yekkanti

Recently I've started looking at RxJSand RxJava(from Netflix) libraries which work on the concept of Reactive Programming.

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

Node.js works on the basis of event loops, which provides you all the arsenal for asynchronous programming and the subsequent node libraries like "cluster" help you to get best out of your multi-core machine. And Node.js also provides you the EventEmitter functionality where you can subscribe to events and act upon it asynchronously.

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

On the other hand if I understand correctly RxJS (and Reactive Programming in general) works on the principle of event streams, subscribing to event streams, transforming the event stream data asynchronously.

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

So, the question is what does using Rx packages in Node.js mean. How different is the Node's event loop, event emitter & subscriptions to the Rx's streams and subscriptions.

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

回答by André Staltz

Observables are not like EventEmitters. They may actlike EventEmitters in some cases, namely when they are multicasted using RxJS Subjects, but usually they don't act like EventEmitters.

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

In short, an RxJS Subjectis like an EventEmitter, but an RxJS Observableis a more generic interface. Observables are more similar to functions with zero arguments.

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

Consider the following:

考虑以下：

---

---

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);

Of course we all expect to see as output:

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

"Hello"
42
"Hello"
42

You can write the same behavior above, but with Observables:

您可以编写上述相同的行为，但使用 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);
});

And the output is the same:

输出是一样的：

"Hello"
42
"Hello"
42

That's because both functions and Observables are lazy computations. If you don't call the function, the console.log('Hello')won't happen. Also with Observables, if you don't "call" (subscribe), the console.log('Hello')won't happen. Plus, "calling" or "subscribing" is an independent operation: two function calls trigger two separate side effects, and two Observable subscribes trigger two separate side effects. As opposed to EventEmitters which share the side effects and have eager execution regardless of the existence of subscribers, Observables have no shared execution and are lazy.

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

---

---

So far, no difference between the behavior of a function and an Observable. This StackOverflow question would have been better phrased as "RxJS Observables vs functions?".

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

Some people claim that Observables are asynchronous. That is not true. If you surround a function call with logs, like this:

有些人声称 Observables 是异步的。那不是真的。如果你用日志包围一个函数调用，像这样：

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

You will obviously see the output:

您显然会看到输出：

"before"
"Hello"
42
"after"

And this is the same behavior with Observables:

这与 Observables 的行为相同：

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

And the output:

和输出：

"before"
"Hello"
42
"after"

Which proves the subscription of foowas entirely synchronous, just like a function.

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

---

---

So what is really the difference between an Observable and a function?

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

Observables can "return" multiple values over time, something which functions cannot. You can't do this:

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

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

Functions can only return one value. Observables, however, can do this:

函数只能返回一个值。然而，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');

With synchronous output:

带同步输出：

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

But you can also "return" values asynchronously:

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

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);
});

With output:

有输出：

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

---

---

To conclude,

总结一下，

• func.call()means "give me one value immediately (synchronously)"
• obsv.subscribe()means "give me values. Maybe many of them, maybe synchronously, maybe asynchronously"

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

That's how Observables are a generalization of functions (that have no arguments).

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

回答by Sairam Krish

When does a listener gets attached to Emitter ?

侦听器何时附加到 Emitter ？

With event emitters, the listeners are notified whenever an event, that they are interested in happens. When a new listener is added after the event has occurred, he will not know about the past event. Also the new listener will not know the history of events that happened before. Ofcourse, we could manually program our emitter and listener to handle this custom logic.

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

With reactive streams, the subscriber gets the stream of events that happened from the beginning. So the time at which he subscribes is not strict. Now he can perform variety of operations on the stream to get the sub-stream of events that he is interested in.

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

The advantage of this comes out:

这样做的好处就出来了：

• when we need to process the events that happened over time
• order in which they happened
• patterns in which the events happened (Let's say, after every buy event on Google stock, a sell event on Microsoft stock happens within 5 minutes)

• 当我们需要处理随时间发生的事件时
• 它们发生的顺序
• 事件发生的模式（比方说，在谷歌股票的每次买入事件之后，微软股票的卖出事件会在 5 分钟内发生）

Higher order streams:

高阶流：

A Higher-order stream is a "stream of streams": a stream whose event values are themselves streams.

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

With Event emitters, one way of doing it is by having same listener attached to multiple event emitters. It becomes complex when we need to correlate the event happened on different emitters.

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

With reactive streams, it's a breeze. An example from mostjs(which is a reactive programming library, like RxJS but more performant)

使用反应式流，这是轻而易举的。来自mostjs的示例（这是一个反应式编程库，类似于 RxJS 但性能更高）

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

In the above example, we correlate the click events with mouse move events. Deducting patterns across events become so easier to accomplish when events are available as a stream.

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

Having said that, with EventEmitter we could accomplish all this by over engineering our emitters and listeners. It needs over engineering because it is not intended in first place for such scenarios. Whereas reactive streams does this so fluently because it is intended to solve such problems.

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