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

可观察对象不像eventemitter。它们在某些情况下，也就是当它们使用RxJS主题进行多播时，它们的行为可能像eventemitter，但通常它们不像eventemitter。

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

考虑以下内容:

---

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

这是因为函数和可观察对象都是惰性计算。如果你不调用这个函数，console.log('Hello')就不会发生。同样对于可观察对象，如果你不"调用"(subscribe)， console.log('Hello')将不会发生。另外，"调用"或"订阅"是一个独立的操作:两个函数调用会触发两个单独的副作用，两个Observable订阅会触发两个单独的副作用。eventemitter会共享副作用，并且不管订阅者是否存在都有即时执行，而observable没有共享执行，而且是惰性的。

---

到目前为止，函数和Observable的行为没有区别。这个StackOverflow问题应该用"RxJS可观察对象vs函数"来表达。

有些人声称可观察对象是异步的。这不是真的。如果用日志包围函数调用，像这样:

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

您将明显看到输出:

"before"
"Hello"
42
"after"

这和observable的行为是一样的:

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

输出:

"before"
"Hello"
42
"after"

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

---

那么Observable和函数的真正区别是什么呢?

observable可以随时间"返回"多个值，而函数却不能。你不能这样做:

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

函数只能返回一个值。然而，可观察对象可以这样做:

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()表示"给我值"。可能有很多，可能是同步的，也可能是异步的"

这就是为什么可观察对象是函数的泛化(没有参数)。

侦听器何时连接到Emitter

对于事件发射器，每当它们感兴趣的事件发生时，都会通知侦听器。当事件发生后添加新的侦听器时，它将不知道过去的事件。另外，新的听者将不知道以前发生的事件的历史。当然，我们可以手动编程我们的发射器和监听器来处理这个自定义逻辑。

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

这样做的好处是:

• 当我们需要处理随时间发生的事件时
• 发生顺序
• 事件发生的模式(例如，在谷歌股票的每次买入事件之后，微软股票的卖出事件在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，我们可以通过过度设计我们的发射器和监听器来完成所有这些。它需要过多的工程设计，因为它最初并不是为这种场景设计的。而响应式流在这方面做得很流畅，因为它旨在解决这些问题。