RxJS 运算符

尽管 Observable 是基础，但 RxJS 对它的运算符最有用。运算符是使复杂的异步代码易于以声明的方式编写的基本组成部分。

什么是运算符？

运算符就是功能。运算符有两种：

可管道运算符是可以使用语法通过管道传递给 Observables 的类型 observableInstance.pipe(operator())。这些包括filter(...)和 mergeMap(...)。调用时，它们不会更改现有的 Observable 实例。相反，它们返回一个新的 Observable，其订阅逻辑基于第一个 Observable。

管道运算符是一个将 Observable 作为其输入并返回另一个 Observable 的函数。这是一个纯粹的操作：以前的 Observable 保持不变。

管道运算符本质上是一个纯函数，它将一个 Observable 用作输入并生成另一个 Observable 作为输出。订阅输出 Observable 也将订阅输入 Observable。

创建运算符是另一种运算符，可以称为独立函数来创建新的 Observable。例如：of(1, 2, 3)创建一个可观察物体，该物体将依次发射 1、2 和 3。创建运算符将在后面的部分中详细讨论。

例如，被调用的运算符 map 类似于同名的 Array 方法。就像 [1, 2, 3].map(x => x * x)yield一样[1, 4, 9]，Observable 创建如下：

import { of } from 'rxjs';
import { map } from 'rxjs/operators';


map(x => x * x)(of(1, 2, 3)).subscribe((v) => console.log(`value: ${v}`));


// Logs:
// value: 1 
// value: 4
// value: 9

会散发出1，4，9。另一个有用的运算符是 first：

import { of } from 'rxjs';
import { first } from 'rxjs/operators';


first()(of(1, 2, 3)).subscribe((v) => console.log(`value: ${v}`));


// Logs:
// value: 1

请注意，map逻辑上必须动态构建，因为必须为其提供映射功能。相比之下，它 first 可能是一个常数，但是仍然是动态构建的。通常，无论是否需要参数，都构造所有运算符。

管道

Pipeable 运营商的功能，所以他们可以像使用普通的功能：op()(obs)-但在实践中，往往是很多人一起卷积，并迅速成为不可读：op4()(op3()(op2()(op1()(obs))))。因此，Observables 具有一种称为的方法.pipe()，该方法可以完成相同的操作，但更易于阅读：

obs.pipe(
  op1(),
  op2(),
  op3(),
  op3(),
)

从风格上讲 op()(obs)，即使只有一个运算符，也不要使用。obs.pipe(op())是普遍首选的。

创建运算符

什么是创作运算符？与管道运算符不同，创建运算符是可用于创建具有某些常见预定义行为或通过加入其他 Observable 的 Observable 的函数。

创建运算符的典型示例是 interval 函数。它以数字（不是 Observable）作为输入参数，并产生 Observable 作为输出：

import { interval } from 'rxjs';


const observable = interval(1000 /* number of milliseconds */);

在这里查看所有静态创建运算符的列表。

高阶可观察物

可观察对象最通常发出诸如字符串和数字之类的普通值，但令人惊讶的是，经常需要处理可观察对象的可观察对象，即所谓的高阶可观察对象。例如，假设您有一个Observable发射字符串，这些字符串是您想要查看的文件的URL。代码可能看起来像这样：

const fileObservable = urlObservable.pipe(
   map(url => http.get(url)),
);

http.get()为每个单独的 URL 返回一个 Observable（可能是字符串或字符串数组）。现在您有了一个 Observables 的 Observables，一个更高阶的 Observable。

但是如何处理高阶 Observable？通常，通过展平：通过（以某种方式）将高阶 Observable 转换为普通 Observable。例如：

const fileObservable = urlObservable.pipe(
   map(url => http.get(url)),
);

的 concatAll()操作者订阅了各“内部”可观察所散发出来的“外”观察的，和复制所有所发射的值，直到该可观察完成，并继续到下一个。所有值都以这种方式连接在一起。其他有用的扁平化运算符（称为连接运算符）是

• mergeAll() —订阅每个内部 Observable 的到达，然后在到达时发出每个值
• switchAll() —在第一个内部 Observable 到达时订阅它，并在到达时发出每个值，但是在下一个内部Observable到达时，取消订阅前一个，并订阅新值。
• exhaust() —在第一个内部 Observable 到达时订阅它，并在到达时发出每个值，并丢弃所有新到达的内部Observable，直到第一个完成时，然后等待下一个内部Observable。

正如许多阵列库结合 map()和 flat()（或 flatten()）成一个单一的 flatMap()，也有全部的 RxJS 映射当量压扁运营商 concatMap()``mergeMap()``switchMap()，和 exhaustMap()。

大理石图

为了解释操作员的工作方式，文字描述通常是不够的。许多操作员都与时间有关，例如，他们可能以不同的方式延迟，采样，节流或消除反跳值。图通常是一个更好的工具。大理石图是操作员如何工作的直观表示，包括输入的 Observable，操作员及其参数以及输出的 Observable。

在大理石图中，时间向右流动，并且该图描述了如何在 Observable 执行中发出值（“大理石”）。

您可以在下面看到大理石图的解剖图。

在整个文档站点中，我们广泛使用大理石图来说明操作员的工作方式。它们在其他环境中也可能确实有用，例如在白板上，甚至在我们的单元测试中（如ASCII图）。

运营商类别

存在用于不同目的的运算符，它们可以归类为：创建，转换，过滤，联接，多播，错误处理，实用程序等。在下面的列表中，您将找到按类别组织的所有运算符。

有关完整概述，请参见参考页。

创建运算符

• ajax
• bindCallback
• bindNodeCallback
• defer
• empty
• from
• fromEvent
• fromEventPattern
• generate
• interval
• of
• range
• throwError
• timer
• iif

加入创作运营商

这些是 Observable 创建运算符，它们也具有联接功能-发出多个源 Observable 的值。

• combineLatest
• concat
• forkJoin
• merge
• partition
• race
• zip

转型运营商

• buffer
• bufferCount
• bufferTime
• bufferToggle
• bufferWhen
• concatMap
• concatMapTo
• exhaust
• exhaustMap
• expand
• groupBy
• map
• mapTo
• mergeMap
• mergeMapTo
• mergeScan
• pairwise
• partition
• pluck
• scan
• switchMap
• switchMapTo
• window
• windowCount
• windowTime
• windowToggle
• windowWhen

筛选运算符

• audit
• auditTime
• debounce
• debounceTime
• distinct
• distinctKey
• distinctUntilChanged
• distinctUntilKeyChanged
• elementAt
• filter
• first
• ignoreElements
• last
• sample
• sampleTime
• single
• skip
• skipLast
• skipUntil
• skipWhile
• take
• takeLast
• takeUntil
• takeWhile
• throttle
• throttleTime

加盟运营商

另请参见上面的“ 加入创建运算符”部分。

• combineAll
• concatAll
• exhaust
• mergeAll
• startWith
• withLatestFrom

组播运营商

• multicast
• publish
• publishBehavior
• publishLast
• publishReplay
• share

错误处理运算符

• catchError
• retry
• retryWhen

公用事业运营商

• tap
• delay
• delayWhen
• dematerialize
• materialize
• observeOn
• subscribeOn
• timeInterval
• timestamp
• timeout
• timeoutWith
• toArray

条件运算符和布尔运算符

• defaultIfEmpty
• every
• find
• findIndex
• isEmpty

数学运算符和聚合运算符

• count
• max
• min
• reduce

创建自定义运算符

使用该 pipe()函数创建新运算符

如果您的代码中有一个常用的运算符序列，请使用该 pipe()函数将该序列提取到新的运算符中。即使序列不是那么常见，将其分解为单个运算符也可以提高可读性。

例如，您可以创建一个将奇数值丢弃并且将偶数值加倍的函数，如下所示：

import { pipe } from 'rxjs';
import { filter, map } from 'rxjs/operators';


function discardOddDoubleEven() {
  return pipe(
    filter(v => ! (v % 2)),
    map(v => v + v),
  );
}

（该 pipe()功能与.pipe()Observable上的方法类似，但不相同。）

从头开始创建新的运算符

它更复杂，但是如果您必须编写不能由现有运算符的组合构成的运算符（这种情况很少发生），则可以使用 Observable 构造函数从头开始编写运算符，如下所示：

import { Observable } from 'rxjs';


function delay(delayInMillis) {
  return (observable) => new Observable(observer => {
    // this function will called each time this
    // Observable is subscribed to.
    const allTimerIDs = new Set();
    const subscription = observable.subscribe({
      next(value) {
        const timerID = setTimeout(() => {
          observer.next(value);
          allTimerIDs.delete(timerID);
        }, delayInMillis);
        allTimerIDs.add(timerID);
      },
      error(err) {
        observer.error(err);
      },
      complete() {
        observer.complete();
      }
    });
    // the return value is the teardown function,
    // which will be invoked when the new
    // Observable is unsubscribed from.
    return () => {
      subscription.unsubscribe();
      allTimerIDs.forEach(timerID => {
        clearTimeout(timerID);
      });
    }
  });
}

请注意，您必须

1. 实现所有三个观察功能 next()，error()以及 complete()订阅输入可观察的时候。
2. 实现“删除”功能，该功能在 Observable 完成时清除（在这种情况下，通过取消订阅并清除所有待处理的超时）。
3. 从传递给 Observable 构造函数的函数中返回该拆解函数。

当然，这仅是示例。该delay()运营商已经存在。