我认为您也可以通过使用调试器来更好地理解。打开一个 StackBlitz RxJS 项目，创建最简单的示例（取决于您要理解的内容），然后放置一些断点。 AFAIK，使用 StackBlitz，您可以调试 TypeScript 文件，这看起来很棒。
首先，Subject 类extends Observable：
export class Subject<T> extends Observable<T> implements SubscriptionLike { /* ... */ }
现在让我们检查Observable 类。
它有众所周知的pipe method：
pipe(...operations: OperatorFunction<any, any>[]): Observable<any> {
return operations.length ? pipeFromArray(operations)(this) : this;
}
其中pipeFromArray 定义为as follows：
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);
};
}
在阐明上面的 sn-p 中发生了什么之前，重要的是要知道 operators 是。运算符是返回另一个函数的函数，该函数的单个参数是Observable<T>，其返回类型是Observable<R>。有时，T 和 R 可以相同（例如，当使用 filter()、debounceTime()...时）。
例如，map 是defined like this：
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);
};
}
在上面的 sn-p 中，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 中删除列表。