【Python】之 RxPy 操作符总结

创建类操作符

• crate：从头创建一个 Observable，在 observer 方法中检查订阅状态，以便及时停止发射数据或者运算
• defer：只有当订阅者订阅才创建 Observable，为每个订阅创建一个新的 Observable
• from_：把一些对象或数据结构转换为 Observable
• interval：创建一个按照给定的时间间隔发射从 0 开始的整数序列的 Observable
• timer：创建一个在给定的延时之后发射单个数据项的 Observable
• just：将一个对象或对象集转换为发射这个或这些对象的 Observable
• range： 创建一个发射指定范围的整数序列的 Observable
• repeat：创建一个重复发射特定项或序列的 Observable
• start：创建一个发射函数返回值的 Observable
• empty/never/throw：创建一个什么都不做直接通知完成的 Observable、创建一个什么都不做直接通知错误的 Observable，创建一个什么都不做的 Observable

---

转换类操作符

• buffer： 缓存，可以简单的理解为缓存，它定期从 Observable 收集数据到一个集合，然后把这些数据集合打包发射，而不是一次发射一个
• flat_map：扁平映射，将 Observable 发射的数据变换为 Observables 集合，然后将这些 Observable 发射的数据平坦化的放进一个单独的 Observable，可以认为是一个将嵌套的数据结构展开的过程
• group_by：分组，将原来的 Observable 分拆为 Observable 集合，将原始 Observable 发射的数据按 Key 分组，每一个 Observable 发射一组不同的数据
• map：映射，通过对序列的每一项都应用一个函数变换 Observable 发射的数据，实质是对序列中的每一项执行一个函数，函数的参数就是这个数据项
• scan：扫描，对 Observable 发射的每一项数据应用一个函数，然后按顺序依次发射这些值
• window：窗口，定期将来自 Observable 的数据分拆成一些 Observable 窗口，然后发射这些窗口，而不是每次发射一项。类似于 Buffer，但 Buffer 发射的是数据，Window 发射的是 Observable，每一个 Observable 发射原始 Observable 的数据的一个子集

注意：

• flat_map 和 map 是两个非常重要的操作符，map 的操作很简单，就是传入一个函数，这个函数会将数据进行转换，一个输入对应一个输出
• flat_map 和 map 不同，其返回值是一个 Observable，一个输入对应多个输出
• 当转换过程是同步过程时，使用 map，当转换过程是异步过程时使用 flat_map
• Group by 在工作中操作数据库的时候经常用到，就是按某个字段分组，在这里也是相同的意思，会按传递的函数生成的key来分组，注意这里的返回是一个分组的Observable，不能直接订阅，需要再做一次处理

---

过滤类操作符

过滤用于从 Observable 发射的数据中进行选择，其常见操作符如下：

• debounce：—只有在空闲了一段时间后才发射数据，通俗的说，就是如果一段时间没有操作，就执行一次操作
• distinct：去重，过滤重复数据
• element_at：取值，发射某一项数据
• ignore_elements：丢弃所有数据，只发射错误或正常终止的通知
• filter：过滤，仅发射 Observable 中通过检测的项
• first：首项，只发射第一项（或者满足某个条件的第一项）数据
• last：末项，只发射最后一项数据
• sample：取样，定期发射 Observable 最近的数据
• skip：跳过开始的 N 项数据
• skip_last：跳过最后的 N 项数据
• take：从发射的第一项数据开始，只发射后续的 N 项数据
• take_last：只发射最后的N项数据

其中最常用的是 filter，filter 就是过滤，对于数据流，仅发射通过检测的项，有点像 SQL 中的 where 条件，只是这里的条件是一个函数，他会遍历一个个项，并执行这个函数，看是否满足条件，对于 满足条件的才会给到输出流。

---

合并类操作符

合并操作符或者叫组合操作符，其常见如下：

• concat：顺序连接多个 Observable
• merge：将多个 Observable 合并为一个。不同于concat，merge不是按照添加顺序连接，而是按照时间线来连接
• zip：使用一个函数组合多个 Observable 发射的数据集合，然后再发射这个结果。如果多个 Observable 发射的数据量不一样，则以最少的Observable 为标准进行压合
• start_with：在数据序列的开头增加一项数据。start_with 的内部也是调用了 concat
• combine_latest：当两个 Observables 中的任何一个发射了一个数据时，通过一个指定的函数组合每个 Observable 发射的最新数据（一共两个数据），然后发射这个函数的结果
• switch_latest：将 Observable 发射出来的多个 Observables 转换为仅包括最近发射单个项的 Observable
• and/then/when：通过模式(And 条件)和计划(Then 次序)组合两个或多个 Observable 发射的数据集

这里需要注意的是：

• combine_latest类似于 zip，但不同的是，zip 只有在每个Observable都发射了数据才工作，而 combine_latest 任何一个发射了数据都可以工作，每次与另一个 Observable 最近的数据压合
• concat 和 merge 都是合并流，区别在于一个是连接，一个是合并，连接的时候是一个流接另一个流，合并的流是无序的，原来两个流的元素交错，当其中一个结束时，另一个就算是没有结束整个合并过程也会中断

---

其它

• all：判断所有的数据项是否满足某个条件
• amb：给定多个 Observable，只让第一个发射数据的 Observable 发射全部数据，其他 Observable 将会被忽略
• contains：判断在发射的所有数据项中是否包含指定的数据
• default_if_empty：如果原始 Observable 正常终止后仍然没有发射任何数据，就发射一个默认值
• sequence_equal：判断两个 Observable 的数据序列是否相同
• skip_until：丢弃 Observable 发射的数据，直到指定的 Observable 发送数据（丢弃条件数据）
• skip_while：丢弃 Observable 发射的数据，直到一个指定的条件不成立（不丢弃条件数据）
• take_until：当发射的数据满足某个条件后（包含该数据），或者第二个 Observable 发送完毕，终止第一个 Observable 发送数据
• take_while：当发射的数据满足某个条件时（不包含该数据），Observable 终止发送数据