在RxJS中，观察者是否被注入到可观察的执行中？

Question

我已经阅读了ReactiveX留档几次，仍然无法完全理解当观察者订阅可观察文件时会发生什么。

我们来看一个简单的例子：

import { Observable } from 'rxjs'; 

const observable = new Observable(subscriber => {
  subscriber.next(1);
  subscriber.complete();
});

const observer = {
  next: (x) => console.log('got value ' + x),
  error: (err) => console.error('something wrong occurred: ' + err),
  complete: () => console.log('done')
};

observable.subscribe(observer);

StackBlitz代码。

我的问题:

传递给可观察对象的订户对象来自哪里？

从RxJS留档：

可以观察到并不是巧合。subscribe和subscribe在newobservable中（函数subscribe（subscriber）{…}） 具有相同的名称。在图书馆，它们是不同的，但为了实用目的，你可以在概念上平等地考虑它们。

因此，显然传递到可观察构造函数（subscriber）中的subscribe回调的对象实际上不是observer对象。至少如果你按照上面关于库实际工作方式的引述去做的话。

如果传入的不是observer对象，那么subscriber到底是什么。下一步（1）和订阅。complete（）调用？如何连接到观察者中的下一个属性？

澄清编辑：

我知道如何利用RxJS，事实上，人们可以从概念上想象观察者被注入（正如报价所说）。然而，我在这里是想了解它实际上是如何工作的。




Answer 1

匿名用户

不，观察者不会被注入到可观察对象中。

混乱源于这样一个事实，即新的可观察（...）语法更像是一个低级工厂，而不是一个有用的模式。

它或多或少是由更直接的实现所使用的机制，如of（value e1， value e2，...， value eN），from（枚举）和from mEvent（...）。

这些方法是您应该关注的实际用例。

在幕后，所有这些方法都将某种同步或异步价值或交互连接到可观测流的奇妙世界中。为了做到这一点，它们以某种方式表现得像一个合适的观察者：它们生成项目并将其放入流中。为此，他们使用一个名为next的函数。就像Observer实现中的方法一样，bacause实际上是以完全相同的方式调用的。

特别是，您可以在此处查看subscribe方法的实现：

https://github.com/ReactiveX/rxjs/blob/master/src/internal/Observable.ts

如果您想了解订阅期间实际发生的情况，我建议您实际查看代码。但是，在我看来，您应该在熟悉了各种可观察的创建函数之后再尝试。

希望能有帮助。

Answer 2

匿名用户

可观察的创建流程如下：

作者定义了一个可观察的（为了便于解释，这里使用new手动定义）：

const myObservable = new Observable(function subscribe(subscriber) {
  subscriber.next(1);
  subscriber.next(2);
  subscriber.complete();
  return function tearDownLogic() {
    console.log('runs when Observable for whatever reason is done (complete, error, or unsubscribed)')
  }
});

传递给上述可观察构造函数的subscribe回调由可观察构造函数本地保存：

constructor(subscribe?: (this: Observable<T>, subscriber: Subscriber<T>) => TeardownLogic) {
  if (subscribe) {
    this._subscribe = subscribe;
  }
}

因此，我们有一个完整的subscribe函数，由我们或任何其他预制的Observable定义，保存下来供以后执行。

可以通过几种形式之一将观察者传递给subscribe回调。或者，直接作为一到三个函数（next、error、complete），或者作为一个对象使用三种相同方法中的一种或多种。出于本说明的目的，我们将实现最后一个更详细的选项：

const observer = {
  next(v) {
    console.log(v);
  }
  error(err) {
    console.log(err);
  }
  complete() {
    console.log('Observable has now completed and can no longer emit values to observer');
  }
}

现在，有趣的部分开始了。我们将观察者传递给可观察者。订阅（……） 方法：

myObserver.subscribe(observer);

subscribe方法如下所示：

  subscribe(observerOrNext?: PartialObserver<T> | ((value: T) => void),
            error?: (error: any) => void,
            complete?: () => void): Subscription {


    const { operator } = this;
    const sink = toSubscriber(observerOrNext, error, complete);


    if (operator) {
      sink.add(operator.call(sink, this.source));
    } else {
      sink.add(
        this.source || (config.useDeprecatedSynchronousErrorHandling && !sink.syncErrorThrowable) ?
        this._subscribe(sink) :
        this._trySubscribe(sink)
      );
    }


    if (config.useDeprecatedSynchronousErrorHandling) {
      if (sink.syncErrorThrowable) {
        sink.syncErrorThrowable = false;
        if (sink.syncErrorThrown) {
          throw sink.syncErrorValue;
        }
      }
    }


    return sink;
  }

简要介绍了subscribe方法：

1. 以先前讨论过的形式之一接收观测者
2. toSubscriber将观察者转换为Subscriber对象，而不管其传入形式如何（Subscriber实例保存在sink变量中）
3. 注意：运算符变量是未定义的，除非您订阅了运算符。因此，只需忽略operator
周围的 if语句即可
4. Subscriber扩展了Subscription对象（原型链接到该对象），该对象的原型上有两个重要方法：unsubscribe（），add（）
5. 添加（…）用于将“分解逻辑”（函数）添加到可观察的，该逻辑将在可观察的完成或取消订阅时运行。它将接受传递给它的任何函数，将其包装在Subscription对象中，并将该函数放入Subscription的\u unsubscribe变量中。此订阅保存在我们在上面创建的订户上，保存在名为\u subscriptions的变量中。如前所述，我们这样做是为了在订户取消订阅或完成订阅时，执行所有add（）'ed下拉逻辑
6. 作为旁注，可观察到。subscribe（）返回订户实例。因此，您可以调用mySubscriber。在它的任何位置添加（//一些分解逻辑），以添加当可观察的完成或取消订阅时将执行的函数
7. 一个重要的部分现在包含了：this_trySubscribe（sink）运行（作为参数在add（）中）<代码>\u尝试订阅（..）是实际运行subscribe回调的函数，该回调先前由Observable构造函数保存。重要的是，它将sink（我们新的Subscriber实例）作为回调传递给Observable回调。换句话说，当<代码>订户。下一步（1）在可观察的执行中，我们实际上是在接收器中执行next（1）（Subscriber）实例（next（）在Subscriber的原型上）

所以，现在我就到此为止。在toSubscribe内部以及关于取消订阅流程的其他内容中有更多详细信息，但这些内容不在本Q的范围之内

简言之，为了回答标题中的问题，只需将观察者转换为统一的订户对象，即可将其传递到可观察对象。

希望这能在将来帮助其他人。