白筱汐

想都是问题，做都是答案

0%

Rxjs入门指南

发表于 分类于

介绍

Rxjs,JavaScript的响应式扩展库，是一个基于可观察序列的库，用于处理异步事件和编写基于事件驱动的程序。它是
ReactiveX（Reactive Extensions）的 JavaScript 实现之一，提供了丰富的操作符和工具，用于处理和组合异步数据流。

主要概念

以下是 RxJS 的一些重要概念：

  1. Observable（可观察对象）：代表一个异步数据流，可以发送零个或多个值，并在完成或出错时发出信号。Observable
    可以被订阅，以便观察它发出的值或通知。

  2. Observer（观察者）：观察者是一个接口或函数集合，用于处理从 Observable 发出的值、错误和完成通知。它包含了三个方法：next
    用于处理正常的值，error 用于处理错误，complete 用于处理完成通知。

  3. Subscription（订阅）：表示 Observable 和 Observer 之间的连接。订阅可以用于取消 Observable 的执行，释放资源或停止观察者接收值。

  4. Operators（操作符）：用于处理 Observable 发出的数据流的函数。RxJS 提供了许多内置的操作符，如 map、filter、merge、concat
    等，以及许多其他用于数据转换、筛选和组合的操作符。

  5. Subject（主题）：是一个特殊类型的 Observable，它允许将值多路推送给多个观察者。

  6. Schedulers（调度器）：用于控制何时和如何执行 Observable 的操作。RxJS 提供了不同类型的调度器，如 async、queue、asap
    等，可以用来控制异步代码的执行顺序和调度方式。

RxJS 提供了一种响应式编程的范式，使得处理异步数据流变得更加简单和灵活。它在前端开发中被广泛应用于处理用户输入、处理网络请求、管理状态等方面，同时也可以在
Node.js 等后端环境中使用。

简单体验一波

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
<!DOCTYPE html>
<html lang="en">

<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Rxjs Tutorial</title>
    <!--  使用 cdn 的模式  -->
    <script src="https://unpkg.com/rxjs@^7/dist/bundles/rxjs.umd.min.js"></script>
</head>

<body>
<script>
    // 通常你会注册一个事件监听器
    // document.addEventListener('click', () => console.log('Clicked!'));

    // 使用 RxJS 的话，创建一个 observable 来代替
    const {fromEvent} = rxjs

    fromEvent(document, 'click').subscribe(() => console.log('Clicked!'));
</script>
</body>

</html>

Purity (纯净性)

ps：函数式编程里面有个概念叫：纯函数，是指在相同输入的情况下，始终返回相同输出，并且不产生副作用的函数。
具体请查看 函数式编程指南。

RxJS 的强大之处在于它能使用纯函数生成值。这意味着您的代码不易出错。

通常情况下，您会创建一个不纯函数，而您代码中的其他部分可能会扰乱您的状态。

1
2
3
// 非纯函数，副作用会对外部的 count 值造成影响，当外部状态改变时，产生的结果也会改变
let count = 0;
document.addEventListener('click', () => console.log(`Clicked ${++count} times`));

使用 RxJS 可以隔离状态

1
2
3
4
5
6

const {fromEvent, scan} = rxjs;

fromEvent(document, 'click')
    .pipe(scan((count) => count + 1, 0))
    .subscribe((count) => console.log(`Clicked ${count} times`));

scan 扫描操作符的工作原理与数组的 reduce 操作符相同。它获取一个值，并将该值暴露给一个回调函数。回调返回的值将成为下次运行回调时的下一个暴露值。

Flow（流动性）

RxJS 拥有一系列操作符，可帮助您控制事件如何流经 observables。 这就是使用普通 JavaScript 最多允许每秒点击一次的方法。（你也可以理解成节流函数）

1
2
3
4
5
6
7
8
9
let count = 0;
let rate = 1000;
let lastClick = Date.now() - rate;
document.addEventListener('click', () => {
    if (Date.now() - lastClick >= rate) {
        console.log(`Clicked ${++count} times`);
        lastClick = Date.now();
    }
});
1
2
3
4
5
6
7
 const {fromEvent, throttleTime, scan} = rxjs;
fromEvent(document, 'click')
    .pipe(
        throttleTime(1000),
        scan((count) => count + 1, 0)
    )
    .subscribe((count) => console.log(`Clicked ${count} times`));

其他流量控制运算符包括 filter、delay、debounceTime、take、takeUntil、distinct、distinctUntilChanged 等。

Values（值）

对于流经 observables 的值，你可以对其进行转换。
下面的代码展示的是如何累加每次点击的鼠标 x 坐标，先来看使用普通的 JavaScript

1
2
3
4
5
6
7
8
9
10
let count = 0;
const rate = 1000;
let lastClick = Date.now() - rate;
document.addEventListener('click', (event) => {
    if (Date.now() - lastClick >= rate) {
        count += event.clientX;
        console.log(count);
        lastClick = Date.now();
    }
});

使用 RxJS

1
2
3
4
5
6
7
8
9
const {fromEvent, throttleTime, map, scan} = rxjs;

fromEvent(document, 'click')
    .pipe(
        throttleTime(1000),
        map((event) => event.clientX),
        scan((count, clientX) => count + clientX, 0)
    )
    .subscribe((count) => console.log(count));

其他产生值的操作符有 pluck、pairwise、 sample 等等。

Observable (可观察对象)

Observables 是多个值的惰性推送集合，你可以参考下面的表格了解他们的关系。

Single Multiple
Pull Function Iterator
Push Promise Observable

下面是一个 Observable，它在订阅后立即（同步）推送值 1、2、3，并在订阅调用后一秒后推送值 4，然后完成流。

1
2
3
4
5
6
7
8
9
10
const { Observable } = rxjs;
const observable = new Observable((subscriber) => {
   subscriber.next(1);
   subscriber.next(2);
   subscriber.next(3);
   setTimeout(() => {
      subscriber.next(4);
      subscriber.complete();
   }, 1000);
});

要调用 Observable 并查看这些值，我们需要订阅它。

1
2
3
4
5
6
7
8
9
10
11
12
13
console.log('just before subscribe');
observable.subscribe({
   next(x) {
      console.log('got value ' + x);
   },
   error(err) {
      console.error('something wrong occurred: ' + err);
   },
   complete() {
      console.log('done');
   },
});
console.log('just after subscribe');

打印的结果为：

1
2
3
4
5
6
7
just before subscribe
got value 1
got value 2
got value 3
just after subscribe
got value 4
done

关于 拉取 (Pull) vs. 推送 (Push) 大家可以访问官网了解。

RxJS 引入了 Observables，一个新的 JavaScript 推送体系。Observable 是多个值的生产者，并将值“推送”给观察者(消费者)。

Observables 像是没有参数, 但可以泛化为多个值的函数。

Observer (观察者)

观察者是由 Observable 发送的值的消费者。观察者只是一组回调函数的集合，每个回调函数对应一种 Observable 发送的通知类型：next、error 和 complete 。下面的示例是一个典型的观察者对象。

1
2
3
4
5
const observer = {
  next: x => console.log('Observer got a next value: ' + x),
  error: err => console.error('Observer got an error: ' + err),
  complete: () => console.log('Observer got a complete notification'),
};

使用 observer 将其提供给 Observable 的订阅者。

1
observable.subscribe(observer);

当订阅 Observable 时，你也可以提供了一个回调函数作为参数，而不是一个对象。

1
observable.subscribe(x => console.log('Observer got a next value: ' + x));

Subscription (订阅)

Subscription 是表示可清理资源的对象，通常是 Observable 的执行。Subscription 有一个重要的方法，即 unsubscribe，它不需要任何参数，只是用来清理由 Subscription 占用的资源。在上一个版本的 RxJS 中，Subscription 叫做 “Disposable” (可清理对象)。

1
2
3
4
5
6
const { interval } = rxjs;

const observable = interval(1000);
const subscription = observable.subscribe(x => console.log(x));

subscription.unsubscribe();

Subscription 基本上只有一个 unsubscribe() 函数，这个函数用来释放资源或去取消 Observable 执行。

Subscription 还可以合在一起，这样一个 Subscription 调用 unsubscribe() 方法，可能会有多个 Subscription 取消订阅 。你可以通过把一个 Subscription 添加到另一个上面来做这件事：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
const { interval } = rxjs;

const observable1 = interval(400);
const observable2 = interval(300);

const subscription = observable1.subscribe(x => console.log('first: ' + x));
const childSubscription = observable2.subscribe(x => console.log('second: ' + x));

subscription.add(childSubscription);

setTimeout(() => {
  // Unsubscribes BOTH subscription and childSubscription
  subscription.unsubscribe();
}, 1000);

ps: interval 是一个创建操作符号， 返回一个发出无限自增的序列整数, 你可以选择固定的时间间隔进行发送。 第一次并 没有立马去发送, 而是第一个时间段过后才发出。 默认情况下, 这个操作符使用 async 调度器来 提供时间的概念，但也可以给它传递任意调度器。

Subject (主体)

RxJS Subject 是一种特殊类型的 Observable，它允许将值组播给许多观察者。普通的观察对象是单播的（每个订阅的观察者都拥有观察对象的独立执行），而主题则是多播的。

主题（Subject）就像一个可观察对象（Observable），但可以向许多观察者（Observer）进行组播。主体就像事件发射器：它们维护着许多监听者的注册表。

每个主题 （Subject）既是一个可观察对象（Observable），又是一个观察者（Observer）。

请查看实例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
const { Subject } = rxjs;
const subject = new Subject();
 
subject.subscribe({
  next: (v) => console.log(`observerA: ${v}`),
});
subject.subscribe({
  next: (v) => console.log(`observerB: ${v}`),
});
 
subject.next(1);
subject.next(2);
 
// Logs:
// observerA: 1
// observerB: 1
// observerA: 2
// observerB: 2

由于每个主题（Subject）也是一个观察者（Observer）,所以我们也可以将一个主题（Subject）作为观察者（Observer）的订阅参数。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
const {Subject, from} = rxjs;

const subject = new Subject();

subject.subscribe({
    next: (v) => console.log(`observerA: ${v}`),
});
subject.subscribe({
    next: (v) => console.log(`observerB: ${v}`),
});

const observable = from([1, 2, 3]);

observable.subscribe(subject);

// Logs:
// observerA: 1
// observerB: 1
// observerA: 2
// observerB: 2
// observerA: 3
// observerB: 3

“多播 Observable” 通过 Subject 来发送通知，这个 Subject 可能有多个订阅者，然而普通的 “单播 Observable” 只发送通知给单个观察者。

多播 Observable 在底层是通过使用 Subject 使得多个观察者可以看见同一个 Observable 执行。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
const {from, Subject, multicast} = rxjs;

const source = from([1, 2, 3]);
const subject = new Subject();
const multicasted = source.pipe(multicast(subject));

// These are, under the hood, `subject.subscribe({...})`:
multicasted.subscribe({
    next: (v) => console.log(`observerA: ${v}`),
});
multicasted.subscribe({
    next: (v) => console.log(`observerB: ${v}`),
});

// 在底层使用了 `source.subscribe(subject)`:
multicasted.connect();

关于 refCount()、ReplaySubject、AsyncSubject、Void subject 请大家查看官网了解。

Operators (操作符)

尽管 RxJS 的根基是 Observable，但最有用的还是它的操作符。操作符是允许复杂的异步代码以声明式的方式进行轻松组合的基础代码单元。

Pipeable Operator（管道操作符）是一个函数，它将一个 Observable 作为输入，并返回另一个 Observable。它是一种纯粹的操作：前一个 Observable 保持不变。

Creation Operators （创建操作符）是另一种操作符，可以作为独立函数调用，创建一个新的 Observable 。

// 使用 of 创建 observable, 使用 map 对数据进行处理。

1
2
3
4
5
6
7
8
const { of, map } = rxjs;

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

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

通过 Pipe（管道）可以将多个操作符放在一起使用, 这段代码我们在前面已经见识到了，现在可以理解它了。

1
2
3
4
5
fromEvent(document, 'click').pipe(
        throttleTime(1000), // 节流 每 1s 只能触发一次 
        map((event) => event.clientX), // 转换数据，每次点击发送鼠标的 clientX
       scan((count, clientX) => count + clientX, 0) // 对数据进行累加，类似 js array reduce
    ).subscribe((count) => console.log(count));

要解析操作符是如何工作的，请查看 [Marble diagrams](Marble diagrams)。

操作符有着不同的用途，它们可作如下分类：创建、转换、过滤、组合、错误处理、工具，等等。

Scheduler (调度器)

调度器控制着何时启动 subscription 和何时发送通知。它由三部分组成：

调度器是一种数据结构， 它知道如何根据优先级或其他标准来存储任务和将任务进行排序。
调度器是执行上下文，它表示在何时何地执行任务(举例来说，立即的，或另一种回调函数机制(比如 setTimeout 或 process.nextTick)，或动画帧)。
调度器有一个(虚拟的)时钟，调度器功能通过它的 getter 方法 now() 提供了“时间”的概念。在具体调度器上安排的任务将严格遵循该时钟所表示的时间。

在下面的示例中，我们使用通常的简单 Observable 同步发送值 1、2、3，并使用操作符 observeOn 来指定使用异步调度器来发送这些值。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
const { Observable, observeOn, asyncScheduler } = rxjs;

const observable = new Observable((observer) => {
  observer.next(1);
  observer.next(2);
  observer.next(3);
  observer.complete();
}).pipe(
  observeOn(asyncScheduler)
);

console.log('just before subscribe');
observable.subscribe({
  next(x) {
    console.log('got value ' + x);
  },
  error(err) {
    console.error('something wrong occurred: ' + err);
  },
  complete() {
    console.log('done');
  },
});
console.log('just after subscribe');

输出结果：

1
2
3
4
5
6
just before subscribe
just after subscribe
got value 1
got value 2
got value 3
done

异步调度程序使用 setTimeout 或 setInterval 进行操作，即使给定的延迟为零。在 js 的事件循环中，异步消息队列在同步代码执行之后执行。

具体调度器的类型和使用指南请参考官网 scheduler。

ps:我了解到 Rxjs，它类似于一个事件流，因为我学习 Dart 的时候使用了 Stream 以及 listen、StreamSubscription。