操作符

Operator（操作符）

尽管RxJS的根基是Observable，但最有用的还是它的操作符。操作符是允许复杂的异步代码以声明式的方式进行轻松组合的基础代码单元。操作符是Observable类型上的方法，比如.map(…)、.filter(…)、.merge(…)，等等。当操作符被调用时，它们不会改变已经存在的Observable实例。相反，它们返回一个新的Observable，它的subscription逻辑基于第一个Observable 。

操作符是函数，它基于当前的Observable创建一个新的Observable。这是一个无副作用的操作：前面的Observable保持不变。操作符本质上是一个纯函数(pure function)，它接收一个Observable作为输入，并生成一个新的Observable作为输出。订阅输出Observable同样会订阅输入Observable 。在下面的示例中，我们创建一个自定义操作符函数，它将从输入Observable接收的每个值都乘以10：

function multiplyByTen(input) {
  const output = Rx.Observable.create(function subscribe(observer) {
    input.subscribe({
      next: (v) => observer.next(10 * v),
      error: (err) => observer.error(err),
      complete: () => observer.complete(),
    });
  });
  return output;
}

const input = Rx.Observable.from([1, 2, 3, 4]);
const output = multiplyByTen(input);
output.subscribe((x) => console.log(x));

实例操作符vs.静态操作符

通常提到操作符时，我们指的是实例操作符，它是Observable实例上的方法。举例来说，如果上面的multiplyByTen是官方提供的实例操作符，它看起来大致是这个样子的：

Rx.Observable.prototype.multiplyByTen = function multiplyByTen() {
  const input = this;
  return Rx.Observable.create(function subscribe(observer) {
    input.subscribe({
      next: (v) => observer.next(10 * v),
      error: (err) => observer.error(err),
      complete: () => observer.complete(),
    });
  });
};

注意，这里的input Observable不再是一个函数参数，它现在是this对象。下面是我们如何使用这样的实例运算符：

const observable = Rx.Observable.from([1, 2, 3, 4]).multiplyByTen();

observable.subscribe((x) => console.log(x));

除了实例操作符，还有静态操作符，它们是直接附加到Observable类上的。静态操作符在内部不使用this关键字，而是完全依赖于它的参数。最常用的静态操作符类型是所谓的创建操作符。它们只接收非Observable参数，比如数字，然后创建一个新的Observable，而不是将一个输入Observable转换为输出Observable 。一个典型的静态操作符例子就是interval函数。它接收一个数字(非Observable)作为参数，并生产一个Observable作为输出：

const observable = Rx.Observable.interval(1000 /* 毫秒数 */);

Marble diagrams (弹珠图)

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

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

操作符案例

控制流动

// 输入 "hello world"
const input = Rx.Observable.fromEvent(document.querySelector("input"), "input");

// 过滤掉小于3个字符长度的目标值
input
  .filter((event) => event.target.value.length > 2)
  .map((event) => event.target.value)
  .subscribe((value) => console.log(value)); // "hel"

// 延迟事件
input
  .delay(200)
  .map((event) => event.target.value)
  .subscribe((value) => console.log(value)); // "h" -200ms-> "e" -200ms-> "l" ...

// 每200ms只能通过一个事件
input
  .throttleTime(200)
  .map((event) => event.target.value)
  .subscribe((value) => console.log(value)); // "h" -200ms-> "w"

// 停止输入后200ms方能通过最新的那个事件
input
  .debounceTime(200)
  .map((event) => event.target.value)
  .subscribe((value) => console.log(value)); // "o" -200ms-> "d"

// 在3次事件后停止事件流
input
  .take(3)
  .map((event) => event.target.value)
  .subscribe((value) => console.log(value)); // "hel"

// 直到其他 observable 触发事件才停止事件流
const stopStream = Rx.Observable.fromEvent(
  document.querySelector("button"),
  "click"
);
input
  .takeUntil(stopStream)
  .map((event) => event.target.value)
  .subscribe((value) => console.log(value)); // "hello" (点击才能看到)

产生值

// 输入 "hello world"
const input = Rx.Observable.fromEvent(document.querySelector("input"), "input");

// 传递一个新的值
input
  .map((event) => event.target.value)
  .subscribe((value) => console.log(value)); // "h"

// 通过提取属性传递一个新的值
input.pluck("target", "value").subscribe((value) => console.log(value)); // "h"

// 传递之前的两个值
input
  .pluck("target", "value")
  .pairwise()
  .subscribe((value) => console.log(value)); // ["h", "he"]

// 只会通过唯一的值
input
  .pluck("data")
  .distinct()
  .subscribe((value) => console.log(value)); // "helo wrd"

// 不会传递重复的值
input
  .pluck("data")
  .distinctUntilChanged()
  .subscribe((value) => console.log(value)); // "helo world"

最近更新于0001-01-01