可观察的主题事件侦听器

我认为您也可以通过使用调试器来更好地理解。打开一个StackBlitz RxJS项目，创建最简单的示例（取决于您要理解的内容），然后放置一些断点。AFAIK，使用StackBlitz可以调试TypeScript文件，这看起来很棒。

---

首先，Subject该类扩展Observable：

export class Subject<T> extends Observable<T> implements SubscriptionLike { /* ... */ }

现在让我们检查一下Observable课程。

它具有众所周知的pipe方法：

pipe(...operations: OperatorFunction<any, any>[]): Observable<any> {
  return operations.length ? pipeFromArray(operations)(this) : this;
}

其中pipeFromArray的定义如下：

export function pipeFromArray<T, R>(fns: Array<UnaryFunction<T, R>>): UnaryFunction<T, R> {
  if (fns.length === 0) {
    return identity as UnaryFunction<any, any>;
  }

  if (fns.length === 1) {
    return fns[0];
  }

  return function piped(input: T): R {
    return fns.reduce((prev: any, fn: UnaryFunction<T, R>) => fn(prev), input as any);
  };
}

在弄清上面片段中发生了什么之前，重要的是要知道运算符。运算符是一个函数，它返回另一个函数，该函数的单个参数是an Observable<T>，返回类型是an Observable<R>。有时候，T并且R可以是相同的（例如，使用时filter()，debounceTime()...）。

例如，map被定义如下：

export function map<T, R>(project: (value: T, index: number) => R, thisArg?: any): OperatorFunction<T, R> {
  return operate((source, subscriber) => {
    // The index of the value from the source. Used with projection.
    let index = 0;
    // Subscribe to the source, all errors and completions are sent along
    // to the consumer.
    source.subscribe(
      new OperatorSubscriber(subscriber, (value: T) => {
        // Call the projection function with the appropriate this context,
        // and send the resulting value to the consumer.
        subscriber.next(project.call(thisArg, value, index++));
      })
    );
  });
}

export function operate<T, R>(
  init: (liftedSource: Observable<T>, subscriber: Subscriber<R>) => (() => void) | void
): OperatorFunction<T, R> {
  return (source: Observable<T>) => {
    if (hasLift(source)) {
      return source.lift(function (this: Subscriber<R>, liftedSource: Observable<T>) {
        try {
          return init(liftedSource, this);
        } catch (err) {
          this.error(err);
        }
      });
    }
    throw new TypeError('Unable to lift unknown Observable type');
  };
}

因此，operate将返回一个函数。请注意其参数：source: Observable<T>。返回类型源自Subscriber<R>。

Observable.lift只是创建一个新的Observable。这就像在喜欢的列表中创建节点。

protected lift<R>(operator?: Operator<T, R>): Observable<R> {
  const observable = new Observable<R>();
  
  // it's important to keep track of the source !
  observable.source = this;
  observable.operator = operator;
  return observable;
}

因此，运算符（如map）将返回一个函数。调用该函数的是pipeFromArray函数：

export function pipeFromArray<T, R>(fns: Array<UnaryFunction<T, R>>): UnaryFunction<T, R> {
  if (fns.length === 0) {
    return identity as UnaryFunction<any, any>;
  }

  if (fns.length === 1) {
    return fns[0];
  }

  return function piped(input: T): R {
    // here the functions returned by the operators are being called
    return fns.reduce((prev: any, fn: UnaryFunction<T, R>) => fn(prev), input as any);
  };
}

在上面的代码中，fn有什么operate函数返回：

return (source: Observable<T>) => {
  if (hasLift(source)) { // has `lift` method
    return source.lift(function (this: Subscriber<R>, liftedSource: Observable<T>) {
      try {
        return init(liftedSource, this);
      } catch (err) {
        this.error(err);
      }
    });
  }
  throw new TypeError('Unable to lift unknown Observable type');
};

也许最好还是看看一个例子。我建议您自己尝试使用调试器。

const src$ = new Observable(subscriber => {subscriber.next(1), subscriber.complete()});

该subscriber => {}回调FN将被分配到的Observable._subscribe财产。

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

接下来，让我们尝试添加一个运算符：

const src2$ = src$.pipe(map(num => num ** 2))

在这种情况下，它将从中调用此块pipeFromArray：

// `pipeFromArray`
if (fns.length === 1) {
  return fns[0];
}

// `Observable.pipe`
pipe(...operations: OperatorFunction<any, any>[]): Observable<any> {
  return operations.length ? pipeFromArray(operations)(this) : this;
}

因此，Observable.pipe将调用(source: Observable<T>) => { ... }，这里source是src$ Observable。通过调用该函数（其结果存储在中src2$），它还将调用该Observable.lift方法。

return source.lift(function (this: Subscriber<R>, liftedSource: Observable<T>) {
  try {
    return init(liftedSource, this);
  } catch (err) {
    this.error(err);
  }
});

/* ... */

protected lift<R>(operator?: Operator<T, R>): Observable<R> {
  const observable = new Observable<R>();
  observable.source = this;
  observable.operator = operator;
  return observable;
}

在这一点上，src$是一个Observable实例，它具有的source集合src$和的operator集合function (this: Subscriber<R>, liftedSource: Observable<T>) ...。

在我看来，这全都与链表有关。创建Observable链（通过添加运算符）时，列表从上到下创建。
当尾节点subscribe调用其方法时，将创建另一个列表，这次是从下到上。我喜欢将第一个称为Observable list，将第二个称为Subscribers list。

src2$.subscribe(console.log)

这是在subscribe调用该方法时发生的情况：

const subscriber = isSubscriber(observerOrNext) ? observerOrNext : new SafeSubscriber(observerOrNext, error, complete);
  
  const { operator, source } = this;
  subscriber.add(
    operator
      ? operator.call(subscriber, source)
      : source || config.useDeprecatedSynchronousErrorHandling
      ? this._subscribe(subscriber)
      : this._trySubscribe(subscriber)
  );

  return subscriber;

在这种情况下src2$有一个operator，因此它将调用它。operator定义为：

function (this: Subscriber<R>, liftedSource: Observable<T>) {
  try {
    return init(liftedSource, this);
  } catch (err) {
    this.error(err);
  }
}

位置init取决于所使用的运算符。再次，这里map的init

export function map<T, R>(project: (value: T, index: number) => R, thisArg?: any): OperatorFunction<T, R> {
  return operate( /* THIS IS `init()` */(source, subscriber) => {
    
    // The index of the value from the source. Used with projection.
    let index = 0;
    // Subscribe to the source, all errors and completions are sent along
    // to the consumer.
    source.subscribe(
      new OperatorSubscriber(subscriber, (value: T) => {
        // Call the projection function with the appropriate this context,
        // and send the resulting value to the consumer.
        subscriber.next(project.call(thisArg, value, index++));
      })
    );
  });
}

source实际上src$。当source.subscribe()被调用时，它将最终调用提供给回调new Observable(subscriber => { ... })。调用subscriber.next(1)会(value: T) => { ... }从上方调用，然后会调用subscriber.next(project.call(thisArg, value, index++));（project-提供给的回调map）。最后，subscriber.next指console.log。

回到Subject，这_subscribe是调用该方法时发生的情况：

protected _subscribe(subscriber: Subscriber<T>): Subscription {
  this._throwIfClosed(); // if unsubscribed
  this._checkFinalizedStatuses(subscriber); // `error` or `complete` notifications
  return this._innerSubscribe(subscriber);
}

protected _innerSubscribe(subscriber: Subscriber<any>) {
  const { hasError, isStopped, observers } = this;
  return hasError || isStopped
    ? EMPTY_SUBSCRIPTION
    : (observers.push(subscriber), new Subscription(() => arrRemove(this.observers, subscriber)));
}

因此，这Subject's就是填充订户列表的方式。通过返回new Subscription(() => arrRemove(this.observers, subscriber))，它可以确保随后的订阅者取消订阅（由于complete/error通知或仅仅是subscriber.unsubscribe()），无效的订阅者将从Subject的列表中删除。