作者：大力智能技术团队-前端 豆花饲养员

写在最前面：

(1) 刚入行时有过一年半Angular的开发经验，当时是做数据看板平台，需要处理看板和图表之间的复杂状态管理，而Angular内置的RxJS数据流管理配合依赖注入的强大特性完美cover住复杂数据流。

(2) 在当下React技术栈开发过程中也使用到RxJS，主要用来做活动流程设计、基础组件数据流管理、跨层级组件通信等，使用数据驱动解耦逻辑。

一、引言

1.1 背景

• 编程范式

命令式编程：通过语句或命令修改程序状态。（eg：JS、Java、Python）

函数式编程：函数是第一等公民，通过函数调用、组合等完成复杂操作。（eg：科里化、函数组合）

声明式编程：描述程序逻辑而非控制流。（eg：SQL、HTML、CSS）

响应式编程：基于数据流及其变化传播的声明性编程范式。（eg：ReactUI = f(state)）

Rx (Reactive Extensions)：响应式扩展 RxJS（Reactive Extensions for JavaScript）：JS的响应式扩展

• RxJS能做什么？

Think of RxJS as Lodash for events.

将同步/异步操作统一抽象到流，关注数据流的变更，提供对数据流的一系列操作和处理方法。通过数据流驱动，有效降低代码逻辑耦合。

1.2 前置概念

1.2.1 流

In computer science, a stream is a sequence of data elements made available over time.

A stream can be thought of as items on a conveyor belt being processed one at a time rather than in large batches.

流是一系列随时间到达的数据。例如：事件流、直播数据流、文本编辑流、WebSocket。

流可抹平同步和异步差异，异步和同步数据都可放入流中。

1.2.2 观察者模式

定义对象间的一种一对多依赖关系，使得每当一个对象状态发生改变时，其相关依赖对象皆得到通知并被自动更新。

✅优点

• 抽象耦合，可实现UI层和数据层分离；
• 抽象数据接口，定义稳定的消息传播机制；
• 支持1对多广播通信

❌缺点

• 观察层级较深时，广播通信成本高；
• 观察者和目标之间存在循环依赖时，将导致死循环崩溃；
• 观察者仅感知目标变化，不感知其知变化过程

二、核心概念

2.1 Observable & Observer

1. Observable

Observables are lazy Push collections of multiple values.

Observables是多个值的懒推送集合。

• 多个值：可推送多个值
• 懒推送：在被第一次订阅时才推送值

2. Pull vs Push

Pull: 数据消费者主动获取数据，数据生产方无感知。

Push：数据生产方决定发送数据给消费者的时机，数据消费者无感知。

• Function 调用时同步返回单个值；
• Iterator 迭代式调用，调用时返回多个值；
• Promise 异步返回单个值；
• Observable 同步/异步地返回多个值。

3. Observer

An Observer is a consumer of values delivered by an Observable.

Observer是Observable的数据消费者。

Observer就是普通对象，类型定义如下：

interface Observer<T> {
    /** 是否已关闭 */
    closed?: boolean;
    /** 流发出值时执行 */
    next: (value: T) => void;
    /** 流出错时执行 */
    error: (err: any) => void;
    /** 流完成时执行 */
    complete: () => void;
}

// 订阅流
// observable.subscribe(observer);

4. Subscription

A Subscription is an object that represents a disposable resource, usually the execution of an Observable.

订阅关系代表一个可关闭的资源（通常是Observable的执行）。

// 订阅流
const subscription = stream$.subscribe({
    next: (val) => console.info,
    error: console.error,
    complete: () => console.info('completed'),
});

// 清理订阅
subscription.unsubscribe();

❗️❗️❗️订阅完成后必须调用unsubscribe()以释放资源，防止内存泄露。

2.2 Subject

A Subject is like an Observable, but can multicast to many Observers. Subjects are like EventEmitters: they maintain a registry of many listeners.

Subject是一种支持多播的Observable。

• 既是Observable：可以直接订阅以获取数据
• 也是Observer：可以调用next(v)、error(e)、complete()方法，将被广播至所有注册的Observer。

Subject类型	特点	示意图
Subject	* 普通Subject * 支持多播
BehaviorSubject	* 随时间变化的数据流 * 缓存当前值 * 被订阅时会立即推送最新值
ReplaySubject	* 缓存最近的若干值 * 可设置缓存窗口
AsyncSubject	* 仅当完成时发出值

2.3 Operator

Operators are the essential pieces that allow complex asynchronous code to be easily composed in a declarative manner.

操作符是声明式处理复杂异步的重要基石。

📌 RxJS ships with operators. —— Ben Lesh

RxJS中操作符都是纯函数。官方提供的操作符基本上够日常开发使用，如果有特殊需求可以自定义操作符。

1. 管道操作符

接收源Observable作为参数，返回新Observable。纯操作，即不改变源Observable。

管道操作符类别参考👉。

2. 创建操作符

用来创建Observable的函数，如

• of：按顺序发出任意数量的值

• interval：基于给定时间间隔发出数字序列

• fromEvent：将DOM事件转化为Observable
• fromPromise：将Promise转化为Observable
• ...

2.4 Scheduler

Scheduler controls when a subscription starts and when notifications are delivered.

调度器控制订阅开始时机、流的数据传播时机。

1. 调度器提供：

• 数据存储和消息队列调度
• 订阅执行上下文
• 虚拟时钟

2. 调度器类别

💡在大部分场景下，使用默认调度器即可。

三、简单实践

3.1 拖拽小球

StackBlitz项目rxjs-drag

1. 拖拽过程

小球一次拖拽过程可描述为：

• 鼠标按下(mousedown)：拖拽开始，记录鼠标按下位置
• 鼠标移动(mousemove)：记录小球初始位置，在鼠标移动时更新小球位置
• 鼠标弹起(mouseup)：本次拖拽结束

2. 数据流

拖拽过程中触发的鼠标事件对应三个流：鼠标按下mousedown$、鼠标移动mousemove$、鼠标弹起mouseup$。

拖拽流drag$可描述如下：

• mousedown$触发拖拽，记录鼠标按下位置
• 切换(switchMap)至mousemove$，记录小球初始位置，并在移动过程中计算小球移动位置
• 当mouseup$触发时，终止(takeUntil)mousemove$

const mousedown$ = fromEvent<MouseEvent>(ball, 'mousedown').pipe(
  map(getMouseEventPos)
);

const mousemove$ = fromEvent<MouseEvent>(document, 'mousemove').pipe(
  map(getMouseEventPos)
);

const mouseup$ = fromEvent<MouseEvent>(document, 'mouseup');

const drag$ = mousedown$.pipe(
  switchMap(initialPos => {
    const { top, left } = ball.getBoundingClientRect();

    return mousemove$.pipe(
      map(({ x, y }) => ({
        top: y - initialPos.y + top,
        left: x - initialPos.x + left
      })),
      takeUntil(mouseup$)
    );
  })
);

小球订阅drag$，更新自身位置。

drag$.subscribe(({ top, left }) => {
  ball.style.top = `${top}px`;
  ball.style.left = `${left}px`;
  ball.style.bottom = '';
  ball.style.right = '';
});

3.2 toast管理

StackBlitz项目rxjs-toast

1. toast需求

一次toast可描述如下：（约定toast停留时长为3s，且不支持手动隐藏）

• 展示一段文本
• 3s内无新toast出现，自动隐藏
• 3s内有新toast出现，重新计时3s，并覆盖老toast

2. 数据流

将toast管理拆分为：

• 展示show$：触发展示
• 隐藏hide$：到达停留时长，或新toast覆盖当前

const click$ = fromEvent(document.getElementById('btn'), 'click');
const toast$ = click$.pipe(
  switchMap(() => {
    let hideByDuration = false;

    const duration$ = timer(2000).pipe(
      mapTo('hide by duration'),
      tap(() => (hideByDuration = true))
    );

    return concat(of('show'), duration$).pipe(
      finalize(() => {
        if (!hideByDuration) {
          console.log('hide by next');
        }
      })
    );
  })
);

toast$.subscribe(console.info);

四、结语

RxJS在流程控制、多个异步协作等场景中均表现良好，使用RxJS写出的代码简洁高效。希望大家在技术选型时可多多考虑RxJS。

参考文档

• ReactiveX
• RxJS官网
• 学习RxJS操作符
• 图解设计模式