关于javascript:RxJS-源码解析四-Operator-II

DevUI是一支兼具设计视角和工程视角的团队，服务于华为云DevCloud平台和华为外部数个中后盾零碎，服务于设计师和前端工程师。
官方网站：devui.design
Ng组件库：ng-devui（欢送Star）
官网交换群：增加DevUI小助手（微信号：devui-official）进群
DevUIHelper插件：DevUIHelper-LSP（欢送Star）

本篇是 RxJS 源码解析的第四篇文章，应用源码的版本是 6.6.0 。本篇文章的内容依然可能会比拟多，请急躁浏览。为了不便浏览，文中的相干代码均通过裁剪和解决。如有不妥，还请斧正。

在本文开始之前，先定义一些自定义术语，不便浏览。

顶流：调用了操作符的流。
上游流：操作符的外部订阅器所订阅的流。
上游流：由操作符的外部订阅器治理的流。
终结订阅：订阅了操作符生成的流的订阅者。

我并不打算像上一篇那样，抓着几个操作符一顿输入。从这篇开始，无论是 Join Operator、还是 Transformation Operator，都有很大的规律性。所以我想先总结进去它们的法则，再来对 operator 进行剖析。

如何管制上游流

为了让操作符能够管制上游流，RxJS 通过委托模式，让操作符生成的了一个特定的 Subscriber，它在外部就能拿到所有传入的上游流的订阅。因而，在这里先介绍两个 Subscriber： OuterSubscriber 和 InnerSubscriber 。

OuterSubscriber ：相当于委托者，提供了三个 notify 的接口—— notifyNext， notifyComplete， notifyError 。
InnerSubscriber ：相当于被委托者，在它结构的时候须要传入 OuterSubscriber ，之后触发绝对应的订阅操作，它会去调用 OuterSubscriber 绝对应的 notify 接口。

其外部实现实际上就是把 InnerSubscriber 的 next，error，complete 转发给 OuterSubscriber 。

export class InnerSubscriber<T, R> extends Subscriber<R> {  private index = 0;  constructor(private parent: OuterSubscriber<T, R>, public outerValue: T, public outerIndex: number) {    super();  }  protected _next(value: R): void {    this.parent.notifyNext(this.outerValue, value, this.outerIndex, this.index++, this);  }  protected _error(error: any): void {    this.parent.notifyError(error, this);    this.unsubscribe();  }  protected _complete(): void {    this.parent.notifyComplete(this);    this.unsubscribe();  }}

而 OuterSubscriber 的默认实现则是将数据交由终结订阅转发进来。

export class OuterSubscriber<T, R> extends Subscriber<T> {  notifyNext(outerValue: T, innerValue: R,             outerIndex: number, innerIndex: number,             innerSub: InnerSubscriber<T, R>): void {    this.destination.next(innerValue);  }  notifyError(error: any, innerSub: InnerSubscriber<T, R>): void {    this.destination.error(error);  }  notifyComplete(innerSub: InnerSubscriber<T, R>): void {    this.destination.complete();  }}

不同的操作符可能会要生成不同的 Subscriber，而生成这些 Subscriber 都会调用 subscribeToResult。这个函数会依据传入的 ObservableInput ，进行类型判断，并返回一个正确处理后的订阅。这里为了能够复用，就调用了之前 from 也应用过的 subscribeTo ，在这个函数中，会解决列表、Promise、以及生成器等数据并返回一个订阅。

export function subscribeToResult<T, R>(  outerSubscriber: OuterSubscriber<T, R>,  result: any,  outerValue?: T,  outerIndex: number = 0,  innerSubscriber: Subscriber<R> = new InnerSubscriber(outerSubscriber, outerValue, outerIndex)): Subscription | undefined {  if (innerSubscriber.closed) {    return undefined;  }  if (result instanceof Observable) {    return result.subscribe(innerSubscriber);  }  return subscribeTo(result)(innerSubscriber) as Subscription;}

通过这种设计，使得生成的 Subscriber 领有管制上游流状态的能力。这种能力能够使得数据装箱和拆箱都放在同一个 Subscriber 中，同时这样做也把副作用集中在一个订阅器中解决，使得操作符在体现上像纯函数一样。

上面以及下一篇的内容中，会呈现大量 subscribeToResult ，咱们只须要晓得，这个函数将订阅的数据或信息转发到了 OuterSubscriber 的相干接口中，它的性能不再赘述。

最初，咱们还是要回归到 operators 的源码剖析上。因为整体法则和设计曾经理解结束，那么剖析每一个 operator 的时候，也能通过这些法则来了解某一部分的 operators 为什么要这样设计。

在这里，咱们持续沿着上一篇的内容，先剖析 Join Creation Operators。

race

所谓 race，意味着所有的流都在进行一场赛跑，跑赢的流能够留下并持续发送数据，没跑赢的只能勾销订阅。

const first = interval(1000).pipe(take(1), mapTo('first'));const second = interval(2000).pipe(take(1), mapTo('second'));const race$ = race(first, second);race$.subscribe((v) => console.log(v));// 打印后果// first

race 通过 fromArray 的形式，将输出的 Observable 交由外部订阅器来解决。

export function race<T>(...observables: ObservableInput<any>[]): Observable<T> {  return fromArray(observables).lift(new RaceOperator<T>());}

RaceSubscriber

RaceSubscriber 保留了这么几个状态。

private hasFirst: boolean = false;private observables: Observable<any>[] = [];private subscriptions: Subscription[] = [];

订阅后上游流输入 Observable 会由 observables 缓存起来，而后在上游流输入实现时，对他们进行订阅，并保留订阅对象。

protected _complete() {  const observables = this.observables;  const len = observables.length;  if (len === 0) {    this.destination.complete();  } else {    for (let i = 0; i < len && !this.hasFirst; i++) {      let observable = observables[i];      let subscription = subscribeToResult(this, observable, observable as any, i);      if (this.subscriptions) {        this.subscriptions.push(subscription);      }      this.add(subscription);    }    this.observables = null;  }}

notify

在 notifyNext 中，RaceSubscriber 能够获取上游流的订阅数据。并对 hasFirst 进行判断。如果该数据是第一个达到，更新 hasFirst 状态，并将其余上游流的订阅勾销，这样做的目标是为了只让这个上游流的数据发送给终结订阅。

notifyNext(  outerValue: T, innerValue: T,  outerIndex: number, innerIndex: number,  innerSub: InnerSubscriber<T, T>): void {  if (!this.hasFirst) {    // 更新状态    this.hasFirst = true;    //     for (let i = 0; i < this.subscriptions.length; i++) {      if (i !== outerIndex) {        let subscription = this.subscriptions[i];        subscription.unsubscribe();        this.remove(subscription);      }    }    this.subscriptions = null;  }  this.destination.next(innerValue);}

zip

zip 是这样的一种操作符，它以上游流中数据量起码的流为基准，依照先后顺序与其余的上游流联合成新的流。

let age$ = of<number>(27, 25, 29, 30, 35, 40);let name$ = of<string>('Foo', 'Bar', 'Beer');let isDev$ = of<boolean>(true, true);zip(age$, name$, isDev$).pipe(  map(([age, name, isDev]) => ({ age, name, isDev })),).subscribe(x => console.log(x));// outputs// { age: 27, name: 'Foo', isDev: true }// { age: 25, name: 'Bar', isDev: true }

zip 也一样，通过 fromArray 的形式，将输出内容交由外部订阅器解决。

export function zip<O extends ObservableInput<any>, R>(  ...observables: O[]): Observable<ObservedValueOf<O>[]|R> {  // 通过 fromArray 将传入的参数以流的模式进入到订阅中  return fromArray(observables, undefined).lift(new ZipOperator());}

ZipSubscriber

订阅开始，生成 ZipSubscriber，调用 _next。依据输出流的类型，将其传入到不同的迭代器中，输出的流的数据类型蕴含了以下几种：

数组
生成器或迭代器
Observable

protected _next(value: any) {  const iterators = this.iterators;  if (isArray(value)) {    iterators.push(new StaticArrayIterator(value));  } else if (typeof value[Symbol_iterator] === 'function') {    iterators.push(new StaticIterator(value[Symbol_iterator]()));  } else {    iterators.push(new ZipBufferIterator(this.destination, this, value));  }}

相较于静态数据而言，Observable 才是咱们关注的重点。在后面曾经讲过 OuterSubscriber 的作用，我在这里不再赘述。 ZipBufferIterator 通过继承 OuterSubscriber，并实现了相应的操作，而后保护了这些 Observable 并进行订阅。

在 zip 中，上游流为 fromArray 生成的 Observable。当它实现时，会把 next 中存储的迭代器进行循环调用。在 next 的时候咱们能够看到，会生成与 ObservableInput 绝对应的内容，的外部如果实现了订阅性能，那么就订阅这些迭代器，否则，间接依照动态解决。

protected _complete() {  const iterators = this.iterators;  const len = iterators.length;  this.unsubscribe();    if (len === 0) {    this.destination.complete();    return;  }    this.active = len;  for (let i = 0; i < len; i++) {    let iterator: ZipBufferIterator<any, any> = <any>iterators[i];    if (iterator.stillUnsubscribed) {      const destination = this.destination as Subscription;      // 持有并治理该迭代器的订阅      destination.add(iterator.subscribe(iterator, i));    } else {      // 不是 Observable      this.active--;    }  }}

ZipBufferIterator 继承了 OuterSubscriber ，那么它必定也是通过外部保护一个 InnerSubscriber 来将上游流中的数据转发进来。

class ZipBufferIterator<T, R> extends OuterSubscriber<T, R> implements LookAheadIterator<T> {  ...  subscribe(value: any, index: number) {    const subscriber = new InnerSubscriber(this, index as any, undefined);    return subscribeToResult<any, any>(this, this.observable, undefined, undefined, subscriber);  }  ...}

notify

ZipBufferIterator 其外部保护了 InnerSubscriber ，那么意味着数据会由发送到 notifyNext 中，这里应用了一个数组将数据缓存起来。

notifyNext(outerValue: T, innerValue: any,            outerIndex: number, innerIndex: number,            innerSub: InnerSubscriber<T, R>): void {  this.buffer.push(innerValue);  this.parent.checkIterators();}

而后，会调用 ZipSubscriber.checkIterators，这个办法决定了终结订阅的数据起源，同时也给出了终结订阅实现所须要的条件。

checkIterators() {  const iterators = this.iterators;  const len = iterators.length;  const destination = this.destination;  // 是不是所有的迭代器都存在数据。  for (let i = 0; i < len; i++) {    let iterator = iterators[i];    if (typeof iterator.hasValue === 'function' && !iterator.hasValue()) {      return;    }  }  let shouldComplete = false;  // 终结订阅最终拿到的数据  const args: any[] = [];   for (let i = 0; i < len; i++) {    let iterator = iterators[i];    let result = iterator.next();    // 判断迭代器是否曾经实现数据输入    if (iterator.hasCompleted()) {      shouldComplete = true;    }    // 如果后果曾经到了开端，意味着最短的数据曾经输入结束。    // 有可能数据没到开端，然而该迭代器曾经完结。    if (result.done) {      destination.complete();      return;    }            // 收集所有迭代器中的数据。    args.push(result.value);  }  // 发送给终结订阅  destination.next(args);  //   if (shouldComplete) {    destination.complete();  }}

当某一个上游流实现的时候，缓冲区的存在与否会决定终结订阅的是否实现。

notifyComplete() {  if (this.buffer.length > 0) {    this.isComplete = true;    this.parent.notifyInactive();  } else {    this.destination.complete();  }}

如果缓冲区存在数据，那么还得去调用 ZipSubscriber.notifyInactive ，将信息返回给 ZipSubscriber。到了这一步，意味着某一个上游流曾经齐全发送完数据了，那么还得更新 active 的记录。如果 active 最终为 0 ，那么将告诉终结订阅这个流曾经实现了。

notifyInactive() {  this.active--;  if (this.active === 0) {    this.destination.complete();  }}

CombineLatest

跟 zip 不一样，在 CombineLatest 中，每一个上游流的新数据都会和其余上游流的以后的数据相结合，从而造成新的数据并从新的流中转发进来。

export function combineLatest<O extends ObservableInput<any>, R>(  ...observables: O[]): Observable<R> {  return fromArray(observables).lift(new CombineLatestOperator<ObservedValueOf<O>, R>());}export class CombineLatestOperator<T, R> implements Operator<T, R> {  constructor() {}  call(subscriber: Subscriber<R>, source: any): any {    return source.subscribe(new CombineLatestSubscriber());  }}

起始状态跟 zip 一样，也是通过 fromArray 将 ObservableInput 作为上游流的数据输出到 CombineLatestSubscriber 中。把眼光锁定这个 Subscriber，深刻理解一下它的心路历程。

CombineLatestSubscriber

当数据到来的时候，CombineLatestSubscriber 把上游流个体缓存到一个 observables 数组中。

protected _next(observable: any) {  this.values.push(NONE);  this.observables.push(observable);}

当上游流缓存结束的时候，上游流也输入结束，那么便会调用 complete。在这里，complete 做的事件仅仅是将所有的上游流进行订阅，并记录这些流的订阅状态。

protected _complete() {  const observables = this.observables;  const len = observables.length;  if (len === 0) {    this.destination.complete();  } else {    this.active = len;    this.toRespond = len;    for (let i = 0; i < len; i++) {      const observable = observables[i];      const innerSub = new InnerSubscriber(this, observable, i);      this.add(subscribeToResult(this, observable, undefined, undefined, innerSub));    }  }}

在订阅结束所有的上游流后，它们的数据全都会流到 notify 中。

notify

CombineLatestSubscriber 每接管到一个上游流的数据，都会触发 notifyNext。toRespond 记录的是残余未收到数据的上游流的数量，当所有上游流都有数据的时候，那么才会开始联合。

values 通过初始化的索引缓存了每一个上游流以后的数据，当任意一个上游流的数据到来后，都将会更新 values 中对应索引中的缓存数据。

notifyNext(outerValue: T, innerValue: R,            outerIndex: number, innerIndex: number,            innerSub: InnerSubscriber<T, R>): void {  const values = this.values;  const oldVal = values[outerIndex];  let toRespond = 0;  if (this.toRespond) {    // 如果这个数据为NONE，那么则代表以后的    // 上游流是首次发送数据，则 toRespond     // 要减一。    if (oldVal === NONE) {      this.toRespond -= 1;    }    toRespond = this.toRespond;  }  values[outerIndex] = innerValue;  if (toRespond === 0) {    this.destination.next(values.slice());  }}

以上便是 combineLastest 的外围设计。

至于 notifyComplete ，则是解决了以后正在运行的上游流和终结订阅的关系。当 active 缩小到零的时候，意味着须要告诉终结订阅所有数据曾经输入结束了。

notifyComplete(unused: Subscriber<R>): void {  this.active -= 1;  if (this.active === 0) {    this.destination.complete();  }}

forkJoin

相较于 combineLatest ，forkJoin 是一种更为激进的实现。为什么说它激进，因为它判断合并的条件，从上游流有数据输入变成了上游流实现数据输入。它的实现很简略，只须要计算每个完结输入数据的上游流的数量 completed，通过比拟 completed 和上游流总数，就能判断什么时候完结。须要留神的一点，如果所有流都输入了数据，那么 forkJoin 能力把数据发送。

function forkJoinInternal(sources: ObservableInput<any>[], keys: string[] | null): Observable<any> {  return new Observable(subscriber => {    const len = sources.length;    if (len === 0) {      subscriber.complete();      return;    }        const values = new Array(len);    let completed = 0;    let emitted = 0;    // 循环订阅所有的上游流    for (let i = 0; i < len; i++) {      // 将输出转换成 Observable       const source = from(sources[i]);      let hasValue = false;            subscriber.add(source.subscribe({        next: value => {          if (!hasValue) {            hasValue = true;            emitted++;          }          // 记录以后订阅的值          values[i] = value;        },        error: err => subscriber.error(err),        // 解决实现时所须要做的工作        complete: () => {          // 更新上游流订阅实现数          completed++;                    // 判断是否所有的上游流订阅都曾经实现          if (completed === len || !hasValue) {            if (emitted === len) {              // 如果全副的上游流都发送了数据，              // 那么终结订阅将收到所有的上游流              // 的数据。              subscriber.next(values);            }            subscriber.complete();          }        }      }));    }  });}

merge & concat

merge 通过调用 mergeMap 来创立合并流，concat 也是通过 mergeMap 来创立雷同的合并流。这一部分会在下一章讲到。它们两个惟一不同的点就是在于并发的数量上。merge能够并发订阅多个上游流，而 concat 同一时间只能订阅一个上游流。

merge 源码

type Any = ObservableInput<any>;export function merge<T, R>(...observables: Array<ObservableInput<any> | number>): Observable<R> {  let concurrent = Number.POSITIVE_INFINITY;  let last: any = observables[observables.length - 1];  if (typeof last === 'number') {    concurrent = <number>observables.pop();  }  return mergeMap<Any, Any>(x => x, concurrent)(fromArray<any>(observables));}

concat 源码

export function concat1<O extends ObservableInput<any>, R>(...observables: Array<O>): Observable<R> {  return mergeMap<O, O>(x => x, 1)(of(...observables));}

partition

partion 是一种宰割操作，通过传入一个判断函数，使得输入的流一分为二。它通过 filter 来实现，将两个不同的流拆散。其中须要留神的是，第二个 filter 中，传入的是一个求反操作。

export function partition<T>(  predicate: (value: T, index: number) => boolean,  thisArg?: any): UnaryFunction<Observable<T>, [Observable<T>, Observable<T>]> {  return (source: Observable<T>) => [    filter(predicate, thisArg)(source),        // 此处传入的是一个 not，他把整个 predicate 封装。    filter(not(predicate, thisArg) as any)(source)  ] as [Observable<T>, Observable<T>];}

本篇小结

总结一下，本章首先给出一些 operators 的综合法则，而后再对 Join Creation Operators 进行剖析，上面别离用一句话将它们总结一下，就完结本篇的内容。

zip 是以数据量起码的上游流为基准合成的流。
combineLatest 是以数据量最多的上游流为基准合成的流。
forkJoin 以上游流实现的状况作为基准合成的流。
merge & concat 下一章再讲。
partion 将输出流一分为二，造成两个流。

退出咱们

咱们是DevUI团队，欢送来这里和咱们一起打造优雅高效的人机设计/研发体系。招聘邮箱：muyang2@huawei.com。

作者：zcx（公众号：Coder写字的中央）

原文链接：https://mp.weixin.qq.com/s/1b141waT_tAxZR-PZC79kg

往期文章举荐

《RxJS 源码解析（三）—— Operator I》

《Web界面深色模式和主题化开发》

《手把手教你搭建一个灰度公布环境》