design-patterns - 观察者模式设计和 RXJS 库

Question

我是 RXJS 库的新手用户，并试图弄清楚如何正确使用 Observable 和 Subjects。
我正在尝试与模式设计观察者进行比较。
在某些时候，我有一个问题，来自库 RXJS 的 Observable 的实例是否是观察者模式设计的特例？

Answer 1

安 Observable根据定义，它是一个随时间发出数据的实体。这听起来有点含糊，但同时又非常有趣。
在我看来，所有 RxJS 的魔法都是通过 实现的。链表 .
每当您创建 Observable使用 new Observable(subscriber => {}) ，您正在定义源，或 头链表的节点。另外，你有没有想过为什么参数被称为subscriber或 observer ?我也会尝试分享我对此的看法。
主链表是在 Observable.pipe() 的帮助下创建的:

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

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);
  };
}

和 Observable.lift() :

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

如你所知，在 RxJS 中有很多操作符。安 operator是一个返回 的函数另一个功能 其参数是 Observable (类型为 T )并且其返回值也是 Observable (类型 R)。
例如， map() :

export function map<T, R>(project: (value: T, index: number) => R, thisArg?: any): OperatorFunction<T, R> {
  return function mapOperation(source: Observable<T>): Observable<R> {
    if (typeof project !== 'function') {
      throw new TypeError('argument is not a function. Are you looking for `mapTo()`?');
    }
    return lift(source, new MapOperator(project, thisArg));
  };
}

所以，当你有

const src$ = new Observable(s => /* ... */)
  .pipe(
    map(/* ... */)
  )

会发生一些事情:

首先，它将创建 Observable实例;提供的回调(在本例中为 s => ... )将存储在 _subscribe 中属性(property)

pipe()被调用；它将返回 fns[0] ，在本例中为 mapOperation功能

mapOperation将使用 Observable 调用实例作为其参数(来自 pipeFromArray(operations)(this) )；调用时，它会调用 source.lift(new MapOperator(project, thisArg)); ; Observable.lift()是什么将节点添加到这个链表；正如你所看到的，一个节点(除了 HEAD )持有 source和 operator代表它

当您订阅 src$ ，基于此列表，将创建另一个。在这个中，每个节点将是一个 Subscriber .此列表的创建基于以下事实:每个 operator must have a call method

export interface Operator<T, R> {
  call(subscriber: Subscriber<R>, source: any): TeardownLogic;
}

MapOperator is no exception :

export class MapOperator<T, R> implements Operator<T, R> {
  constructor(private project: (value: T, index: number) => R, private thisArg: any) {
  }

  call(subscriber: Subscriber<R>, source: any): any {
    return source.subscribe(new MapSubscriber(subscriber, this.project, this.thisArg));
  }
}

Subscriber之间的关系节点建立在 Observable.subscribe()在这种情况下， s来自 new Observable(s => ...) 的参数(上面的例子)将是 MapSubscriber .
我似乎偏离了这个问题，但通过上述解释，我想证明 Observer 的内容并不多。模式在这里。
这种模式可以通过 Subject 来实现, 其中 extends Observable :

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

这意味着您可以使用 Subject.pipe(...)和 Subject.subscribe(subscriber) .什么 Subject为了实现这个模式确实是有一个自定义 _subscribe 方法:

_subscribe(subscriber: Subscriber<T>): Subscription {
  if (this.closed) {
    throw new ObjectUnsubscribedError();
  } else if (this.hasError) {
    subscriber.error(this.thrownError);
    return Subscription.EMPTY;
  } else if (this.isStopped) {
    subscriber.complete();
    return Subscription.EMPTY;
  } else {
    
    // !!!
    this.observers.push(subscriber);
    return new SubjectSubscription(this, subscriber);
  }
}

如您所见， Subject类跟踪它的观察者(订阅者)，所以当它发出一个值时， Subject.next() ，它的所有观察者都会收到它:

next(value: T) {
  if (this.closed) {
    throw new ObjectUnsubscribedError();
  }
  if (!this.isStopped) {
    const { observers } = this;
    const len = observers.length;
    const copy = observers.slice();
    for (let i = 0; i < len; i++) {
      copy[i].next(value!);
    }
  }
  }

作为侧节点， Subject可以充当 Subscriber同样，因此您不必手动拨打 Subject.{next, error, complete}()每时每刻。你可以用这样的东西来实现

src$.pipe(subjectInstance);