RxJS 操作符概览(RxJS Operators Overview)

作者: billowstao
版权: 知识共享署名 4.0 国际许可协议

RxJS 的操作符太丰富了，总结这一篇 RxJS 的操作符便于快速查询。🤠

• RxJS 操作符概览(RxJS Operators Overview)
  • 弹珠图(Marble Diagrams)
  • 操作符分类(Category Of Operators)
    • 创建操作符(Creation Operators)
    • 连接创建操作符(Join Creation Operators)
    • 转换操作符(Transformation Operators)
    • 过滤操作符(Filtering Operators)
    • 连接操作符(Join Operators, 高阶 Observable 操作符)
    • 多播操作符(Multicasting Operators)
    • 错误处理操作符(Error Handing Operators)
    • 工具操作符(Utility Operators)
    • 条件和布尔操作符(Conditional And Boolean Operators)
    • 数学和集合运算符(Mathematical and Aggregate Operators)
    • 参考(Reference)

弹珠图(Marble Diagrams)

要解释操作符是如何工作的，文字描述通常是不足以描述清楚的。许多操作符都是跟时间相关的，它们可能会以不同的方式延迟(delay)、取样(sample)、节流(throttle)或去抖动(debounce)。图表通常是更适合的工具。弹珠图是操作符运行方式的视觉表示，其中包含输入 Observable(s) (输入可能是多个 Observable)、操作符及其参数和输出 Observable 。

在弹珠图中，时间流向右边，图描述了在 Observable 执行中值(“弹珠”)是如何发出的。

在下图中可以看到解剖过的弹珠图。

在整个文档站中，我们广泛地使用弹珠图来解释操作符的工作方式。它们在其他环境中也可能非常有用，例如在白板上，甚至在我们的单元测试中(如 ASCII 图)。

操作符分类(Category Of Operators)

操作符有着不同的用途，它们可作如下分类：创建、转换、过滤、连接、多播、错误处理、工具，等等。在下面的列表中，你可以按分类组织好的所有操作符。

创建操作符(Creation Operators)

• ajax: 创建 ajax 的 Observable。
• bindCallback: 将回调函数转换为 Observable。
• bindNodeCallback: 将 Node.js 回调函数转换为 Observable(Node.js 的回调函数为 callback(error, result))
• defer: Observable 工厂函数(factory)
• ~~empty: RxJS v8 将要移除。~~使用 EMPTY 或者 scheduled(例如 schedule([], scheduler)) 替换
• from: 从 Array, 类数组对象, Promise, 字面量对象, 类 Observable 对象创建 Observable
• fromEvent: 事件绑定
• fromEventPattern: 事件绑定，事件销毁
• generate: 类似于 for 循环
• interval: 创建指定时间间隔的连续数列
• of: 将参数转换为 Observable 序列
• range: 创建一个 Observable，它会在指定的范围内发出连续的数字
• throwError: 创建返回错误的 Observable (concatMap, mergeMap, defer 不需要调用该方法, 直接 throw 错误对象)
• timer: 延时定时器
• iif: 相当于三元表达式 (断言 ? true : false) 的 Observable

连接创建操作符(Join Creation Operators)

这些是 Observable 的创建操作符，它们也具有连接功能 —— 发出多个源 Observable 的值。

• combineLatest: 连接多个 Observable，并且发出所有输入 Observable 的最近值的结合数组
• concat: 连接不同的 Observable 流(按照 Observable 输入的先后顺序连接)，然后依次发出连接后的每个值
• forkJoin: 按照输入顺序连接不同的 Observable，等待所有输入 Observable 都成功后按照输入顺序发出数组包裹的结果(当且仅当所有输入 Observable 都成功返回值后才返回相应的结果数组)
• ~~merge: RxJS v8 将要移除。~~使用 mergeWith 替代
• mergeWith: 按照输入 Observable 的各自顺序合并所有值为一个新的 Observable
• partition: 按照条件拆分 1 个 Observable 为 2 个 Observable，满足条件的值加入第 1 个返回的 Observable，不满足条件的值加入第 2 个返回的 Observable
• race: 竞赛，输入多个 Observable, 发出最先触发的 Observable
• zip: 组合(打包)多个 Observable，依次发出每个 Observable 值的组合(数组包裹输出的所有输入值；当输入的 Observable 长度不一致，仅输出长度一致的共同部分)

转换操作符(Transformation Operators)

• buffer: 缓冲源 Observable，直到输入 Observable 发出
• bufferCount: 缓冲源 Observable，直到限定数量的源发出
• bufferTime: 缓冲源 Observable，直到限定的时间间隔
• bufferToggle: 从触发 openings 到触发 closingSelector 缓冲源 Observable，并发出缓冲的源 Observable
• bufferWhen: 缓冲源 Observable，直到输入 Observable 触发
• concatMap: 将每个源值映射到一个 Observable 中，并在输出的 Observable 中合并，以序列化的方式等待每个源值都完成，然后再合并下一个(先 map 再 concat，源 Observable 的值保留)
• concatMapTo: 将每个源值映射到同一个 Observable 上，并在输出的 Observable 上以序列化的方式合并多次(源 Observable 的值被丢弃)
• ~~exhaust: 将在 RxJS v8 中移除。~~重命名为 exhaustAll
• exhaustAll: 将一个高阶 Observable 转换为一阶 Observable ，当源内部的前一个 Observable 未完成时丢弃当前的 Observable
• exhaustMap: 将每个源值映射到一个 Observable 中，只有当之前映射的 Observable 完成时，才会合并到输出的 Observable 中
• expand: 递归的将每个值映射到一个被合并到输出 Observable 中的 Observable 中
• groupBy: 将一阶的源 Observable 通过 keySelector 分组成高阶的 Observable 并输出
• map: 将给定的映射函数应用到由源 Observable 发出的每个值上，并将结果值作为 Observable 发出
• mapTo: 每当源 Observable 发出一个值时，就在输出 Observable 上发出给定的常量值
• mergeMap: 将每个源值映射到一个 Observable 中，并合并到输出的 Observable 中
• mergeMapTo: 将每个源值映射到同一个 Observable 中，并在输出 Observable 中多次合并
• mergeScan: 在源 Observable 上应用累加器函数，累加器函数本身返回一个 Observable，然后每个返回的中间 Observable 合并到输出 Observable 中
• pairwise: 将一对对连续的发出组合在一起，并将它们作为两个数组发出
• partition: 按照条件拆分 1 个 Observable 为 2 个 Observable，满足条件的值加入第 1 个返回的 Observable，不满足条件的值加入第 2 个返回的 Observable
• ~~pluck: 将在 RxJS v8 中移除。~~使用 map 替代
• scan: 用于封装和管理状态。在初始状态建立后，对于来自源的每个值应用累加器(或 reducer 函数) – 通过 seed 值(第二个参数)或来自源的第 1 个值
• switchScan: 在源 Observable 上应用累加器函数，累加器函数本身返回一个 Observable，只从最新返回的 Observable 发出值(重叠发出的源 Observable 的值会被最新的值替代)
• switchMap: 将每个源值都映射到一个 Observable，它只从最新映射的 Observable 发出
• switchMapTo: 将每个源值都映射到同一个 Observable，这个 Observable 中使用 switchMap 多次扁平化
• window: 当 windowBoundaries 发出时，分支出源 Observable 的值作为一个被嵌套的 Observable
• windowCount: 分支出源 Observable 的值作为一个嵌套的 Observable，每个嵌套的 Observable 最多发出 windowSize 次数的值
• windowTime: 定时将源 Observable 的值分支出来，作为一个嵌套的 Observable
• windowToggle: 分支出源 Observable 的值作为一个嵌套的 Observable，从 openings 发出开始，到 closingSelector 发出结束
• windowWhen: 分支出源 Observable 的值作为一个嵌套的 Observable，使用用来关闭 Observable 的工厂函数来决定何时启动一个新的窗口

过滤操作符(Filtering Operators)

• audit: 在一段时间内忽略由另一个 Observable 决定的源值，然后从源 Observable 发出最近的值(发出静默时间窗口内的最后一个值)，然后重复这个过程
• auditTime: 忽略 duration 毫秒的源值，然后从源 Observable 发出最近的值，然后重复这个过程
• debounce: 只有由一个由另一个 Observable 决定的特定时间跨度过去而没有另一个 Observable 发出时，才会从源 Observable 发出值
• debounceTime: 只有在一个特定的时间跨度过去之后，没有其他源发出，才会从 Observable 发出
• distinct: 返回一个 Observable，它会发出源 Observable 所发出的所有与之前(发出)的项不同的项
• distinctUntilChanged: 返回一个 Observable，使用 comparator 判断当前项与其他项是否不同而发出
• distinctUntilKeyChanged: 返回一个 Observable，该 Observable 发出源 Observable 发出的所有与前一项不同的项，并使用提供的键(key)访问属性来检查两个项是否不同
• elementAt: 在源 Observable 的一系列发出中，在指定的索引处发出单个值
• filter: 发出源 Observable 满足特定条件的值
• first: 只发出由源 Observable 发出的第一个值(或满足某些条件的第一个值)
• ignoreElements: 忽略由源 Observable 发出的所有项，只传递 complete 或 error 的调用
• last: 返回一个只触发源 Observable 所触发的最后一项的 Observable(可以选择使用一个断言函数作为参数，在这种情况下，生成的 Observable 不会从源 Observable 中发出最后一项，而是从源 Observable 中发出满足断言的最后一项)。
• sample: 当另一个 Observable 发出时，发出源 Observable 中最近发出的值(采样)
• sampleTime: 在周期的时间间隔内从源 Observable 发出最近发出的值
• single: 返回一个 Observable，它判断源 Observable 是否只发出一个与断言匹配的值。如果没有提供断言，那么它将判断 Observable 是否只发出一个值
• skip: 返回一个 Observable，它跳过源 Observable 发出的 count 数量项
• skipLast: 返回源 Observable 跳过完成前 skipCount 数量项
• skipUntil: 发出源 Observable 发出的值，知道 notifier Observable 发出一个值
• skipWhile: 返回一个 Observable，只要指定的条件为 true，它就跳过发出源 Observable 的项，一旦条件变成 false，它就会发出源 Observable 的项
• take: 只发出源 Observable 发出的前 count 数量个值
• takeLast: 只发出源 Observable 发出的最后 count 个值
• takeUntil: 发出源 Observable 发出的值，直到 notifier Observable 发出一个值
• takeWhile: 发出源 Observable 发出的值，只要每个值满足给定的 predicate，然后在这个 predicate 不满足时完成
• throttle: 从源 Observable 发出一个值，然后在由另一个 Observable 决定的持续时间内忽略后续的源值，然后重复这个过程
• throttleTime: 从源 Observable 中发出一个值，然后在持续时间(毫秒)内忽略后续的源值，然后重复这个过程

连接操作符(Join Operators, 高阶 Observable 操作符)

还请参阅上面的连接创建操作符(Join Creation Operators)

• combineLatestAll: 当 Observable 的 Observable 完成时，通过应用 combineLatest 将 Observable 的 Observable 扁平化
• concatAll: 将一个高阶 Observable 转换为一阶 Observable，通过内部 Observable 的顺序连接
• exhaustAll: 将一个高阶 Observable 转换为一阶 Observable，方法是在之前的内部 Observable 尚未完成时，删除当前的内部 Observable
• mergeAll: 将一个高阶 Observable 转换为一阶 Observable，该 Observable 并发地提供了内部 Observable 所发出的所有值
• switchAll: 将高阶 Observable 转换为一阶 Observable，只从最近的 Observable 序列产生值
• startWith: 返回一个 Observable，该 Observable 在进行订阅时将同步发送提供给该操作符的所有值，然后订阅源并将其所有值镜像发送给订阅者
• withLatestFrom: 将源 Observable 与其他 Observable 组合起来创建一个 Observable，其值仅在源发出时根据每个 Observable 的最新值计算

多播操作符(Multicasting Operators)

• ~~multicast: 将在 RxJS v8 中移除。~~使用 connectable Observable, connect 操作符或 share 操作符替代
• connect: 通过在函数中对源进行多播来创建一个 Observable，该函数允许开发人员在连接之前定义多播的用法
• share: 返回一个新的 Observable，它可以多播(共享)原来的 Observable。只要至少有一个订阅者，这个 Observable 就会被订阅并发出数据。当所有的订阅者都取消订阅时，它将取消对源 Observable 的订阅。因为 Observable 是多播的，所以流是 hot(这是 multicast(() => new Subject()), refCount(). 的别名)
• ~~publish: 将要在 RxJS v8 中移除。~~使用 connectable, connect 或 share 替代
• ~~publishBehavior: 要在 RxJS v8 中移除。~~使用 BehaviorSubject 替代
• ~~publishLast: 要在 RxJS v8 中移除。~~使用 AsyncSubject 替代
• ~~publishReplay: 要在 RxJS v8 中移除。~~使用 connectable, connect 或 share 替代

错误处理操作符(Error Handing Operators)

• catchError: 通过返回一个新的 Observable 或抛出一个错误来捕获待处理的 Observable 上的错误
• retry: 当 Observable 出现错误，在一段时间后重新返回源 Observable 镜像的 Observable
• retryWhen: 当 Observable 出现错误返回一个镜像 Observable。如果源 Observable 调用了 error，这个方法将从 notifier 调用 error，这个方法将从 notifier 发出导致这个错误的 Throwable。如果那个可观察对象调用 complete 或 error，那么这个方法就会在子订阅上调用 complete 或 error。否则该方法将重新订阅源 Observable

工具操作符(Utility Operators)

• tap: 用于执行来自源 Observable 发出的副作用
• delay: 将源 Observable 延迟到给定的 timeout 或日期
• delayWhen: 在一个由另一个 Observable 的发出所决定的给定时间跨度内延迟源 Observable 的发出
• dematerialize: 将 ObservableNotification 类型的 Observable 转换为它们表示的数据并发出
• materialize: 使用 Notification 对象标记原始类型，将来自源 Observable 的所有通知转换为 next 发出
• observeOn: 用指定的 scheduler 重新发出源 Observable 中的所有通知
• subscribeOn: 在指定的 SchedulerLike 上异步订阅这个 Observable 的 Observable
• timeInterval: 发出一个对象，该对象包含当前值以及在发出当前值和前一个值之间经过的时间，该时间是通过使用 scheduler 的 now() 方法获取每次发出的当前时间，然后计算时间差。scheduler 默认是 asyncScheduler，默认情况下，interval 默认为毫秒
• timestamp: 将时间戳附加到由 Observable 发出的每个项上，以指示它是何时发出的
• timeout: 如果 Observable 在给定时间间隔内没有发出一个值，则会出现错误
• timeoutWith: 在由另一个 Observable 触发的时间间隔内，如果当前源 Observable 没有发出一个值，则会出现错误
• toArray: 收集源发出的所有数据，并在源完成时将它们作为数组发出

条件和布尔操作符(Conditional And Boolean Operators)

• defaultIfEmpty: 如果源 Observable 完成时没有发出任何一个值，则发出一个给定的值，否则镜像源 Observable
• every: 返回一个 Observable，它发出是否(布尔值)源的每一项都满足指定的表达式
• find: 只发出源 Observable 满足条件的第一个值
• findIndex: 只发出源 Observable 满足条件的第一个值的索引值
• isEmpty: 如果输入 Observable 发出了任何值，则发出 false，如果输入 Observable 完成时没有发出任何值，则发出 true

数学和集合运算符(Mathematical and Aggregate Operators)

• count: 统计源的发出次数，并在源完成时发出该次数
• max: max 操作符对一个 Observable 进行操作，它会发出数字(或者提供可以进行元素比较的函数)，当源 Observable 完成时，它会发出一个元素: 最大的值
• min: min 操作符对一个 Observable 进行操作，它会发出数字(或者提供可以进行元素比较的函数)，当源 Observable 完成时，它会发出一个元素: 最小的值
• reduce: 在源 Observable 上应用一个累加器函数，并在源完成时返回累加的结果，可选给定一个 seed (初始值)

参考(Reference)

RxJS Operators: https://rxjs.dev/guide/operators