在RxJS中，观察者是可观察者的“倾听者”吗？

Question

我正在学习RxJS，对于“听众”在哪里（在可观察的或观察者中），他们是如何订阅/取消订阅的，以及当观察者“不再对”可观察的“不感兴趣”时会发生什么，比如当你使用拿或采取直到。

对于第一部分——什么是订阅什么，什么是倾听者——我对这些陈述之间看似矛盾的地方感到困惑。从http://reactivex.io/rxjs/manual/overview.html我们读到观察者不是观察者的“听众”

这与addEventListener/removeEventListener等事件处理程序API截然不同。可以观察到。订阅时，给定的观察者未注册为可观察对象中的侦听器。Observable甚至不维护附加观察者的列表。

但是在http://reactivex.io/learnrx/它说（练习30）（突出显示我的）

一个基于事件的可观察事物永远不会自行完成。函数的作用是：创建一个新的序列，该序列在离散数量的项目到达后完成。这一点很重要，因为与事件不同，当一个可观察序列完成时，它会取消其所有侦听器的订阅。这意味着，如果我们使用take（）来完成事件序列，就不需要取消订阅！

这对我来说似乎是矛盾的。例如，当您使用fromEvent设置一个可观察对象时，事件侦听器在哪里？例如，在基于DOM事件的可观察对象上使用take（1）时，第一个事件发送给观察者后会发生什么？观察者是否取消了对持续发出事件的可观察者的订阅，只是因为观察者不再听这些事件了？或者被观察者是否以某种方式取消了观察者的订阅，也就是说，eventListener是在被观察者中，而不是在观察者中？

谢谢你提供的任何线索——很明显，我不是在看森林而不是树木，但是我正在学习的教程，虽然它们擅长从概念上解释，但让我对实际发生的事情感到困惑。

Answer 1

第一部分非常详细地介绍了单词的用法，以强调订阅一个可观察对象就是调用一个函数（或者更可能是一个函数链）来运行它们包含的所有代码。第二部分不太讲究措辞，但实际上并不是在谈论同一件事。如果您愿意，第二部分可以更好地表述为“当一个可观察对象完成时，它在其观察对象上调用拆卸逻辑。

让我试着描述一下，当我说订阅一个可观测对象是调用一系列函数的问题时，我的意思是什么。考虑下面的简单例子：

对于一个超级简单的例子，假设我创建了这个可观察的：

const justOne = Rx.Observable.create(function realSubscribe(observer) {
  observer.next(1);
  observer.complete();
});

justOne.subscribe(val => console.log(val));

如果我然后只调用一个。订阅（val）=

在这个过程中，可观察到的没有创建或增加订阅者列表；它只是依次运行代码，然后完成。

现在进入一个更现实的例子，让我们考虑<代码> FuffEng/<代码>。如果我要实现它，它可能看起来像这样（真正的实现更复杂，但这得到了它的要点）：

function fromEvent(element, eventName) {
  return Rx.Observable.create(function subscribeToEvent(observer) {
    element.addEventListener(eventName, observer.next);
    return function cleanup() {
      element.removeEventListener(eventName, observer.next);
    }
  });
}

const observable = fromEvent(document, 'click');
const subscription = observable.subscribe(event => console.log(event));

现在当我称之为可观察的。subscribe，它运行subscribeToEvent，并在这样做时调用文档上的addEventListener。文件addEventListener确实会导致文档保留一个事件侦听器列表，但这是因为addEventListener的实现方式，而不是所有可观察对象的通用方式。可观察对象本身不跟踪任何侦听器。它只是调用它被告知要调用的内容，然后返回一个清理函数。

接下来让我们看看。和以前一样，真正的实现更加复杂，但大致是这样的：

// In the real `take`, you don't need to pass in another observable since that's
// available automatically from the context you called it in. But my sample code
// has to get it somehow.
function take(count, otherObservable) {
  return new Observable(function subscribeToTake(observer) {
    let soFar = 0;
    otherObservable.subscribe((value) => {
      observer.next(value);
      soFar++;
      if (soFar >= count) {
        observer.complete();
      }
    });
  });
}

const clickObservable = fromEvent(document, 'click');
take(1, clickObservable).subscribe(event => console.log(event))

正如评论中提到的，我使用的语法与rxjs中的使用方式不太匹配，但那是因为要模仿它需要更全面的实现。无论如何，要引起你注意的主要事情是我们开始产生一个函数链：

当我调用时，它调用的这将设置一个计数器，然后调用其他Observable.subscribe，该计数器是 然后调用document.addEventListener。

Tue的工作是坐在这个功能链的中间。它跟踪到目前为止已经发出了多少值。如果计数足够低，它只是向前的值。但是一旦达到计数，它将调用完成，从而结束可观察。调用完成会导致可观察到的运行它所拥有的任何拆卸逻辑，或者它的链所拥有的任何东西。对于ake没有拆卸逻辑，但是fromEvent将运行一些拆卸逻辑来删除事件侦听器。