RxJS入门分享

作者: pengji | 来源:发表于2022-07-11 15:45 被阅读0次

RxJS 实战(入门)分享

Reactive Extensions Library for JavaScript

📙 什么是RxJS?

Think of RxJS as Lodash for events.

RxJS 是为响应式编程设计的库，它利用 Observables 模式方便我们编写基于异步组合或者回调的代码)
Rx (Reactive extensions) 利用迭代器和观察者模式（观察者模式、迭代器模式），函数式编程(函数式编程)来优雅的编写和管理事件序列代码的编程思想。

大脑 🧠：懂了，又没完全懂

Tips: 观察者模式 & 迭代器模式

观察者模式
释义：将逻辑分成发布者和观察者。
发布者：负责产生事件，它会通知所有的注册挂号的观察者。不关心观察者如何处理事件。
观察者：只负责接收事件并处理自身逻辑，不关心事件如何产生。
图解：

迭代器模式
释义：指的是一个可遍历的对象数据集合的实现。方式有很多，比如数组、链表、树等等迭代器的作用是就是提供一个统一的遍历接口，不需要使用者关心内部的数据集合的实现方式。
图解：

Tips2：编程范式 - 响应式编程 & 函数式编程

编程范式

命令式编程
声明式编程
事件驱动编程
面向对象(OOP)
函数式编程(FP)
- 含义：函数式编程是种编程方式，它将电脑运算视为函数的计算。函数编程语言最重要的基础是λ演算（lambda calculus），而且λ演算的函数可以接受函数当作输入（参数）和输出（返回值）。
- 特征：
  - 函数是"第一等公民"
  - 闭包和高阶函数
  - 惰性计算
  - 递归
  - 没有"副作用"(side effect)：指的是函数内部与外部互动（最典型的情况，就是修改全局变量的值），产生运算以外的其他结果。函数式编程强调没有"副作用"，意味着函数要保持独立，所有功能就是返回一个新的值，没有其他行为，尤其是不得修改外部变量的值。
响应式编程(RP)
- 响应式编程（英文：Reactive Programming）是一种面向数据流和变更传播的异步编程范式。
- 特点：
  - 异步编程：提供了合适的异步编程模型，能够挖掘多核CPU的能力、提高效率、降低延迟和阻塞等。
  - 数据流：基于数据流模型，响应式编程提供一套统一的Stream风格的数据处理接口.
  - 变化传播：简单来说就是以一个数据流为输入，经过一连串操作转化为另一个数据流，然后分发给各个订阅者的过程。
函数式响应式编程(FRP)
函数式响应式编程(FRP) 是一种编程范式，它采用函数式编程的基础部件（如map、reduce、filter等），进行响应式编程（异步数据流程编程）。

听到了听到了，两只耳朵都听到了，赶快给我说说RxJS

🌰 看个栗子

问1：页面输入框中，如何过滤掉小于3个字符长度的目标值?

/**
 * 常规命令式实现
 */

const input2$ = document.querySelector('.input2');
input2$.addEventListener('input', (event: Event) => {
  const res = (event.target as HTMLInputElement).value;
  if (res.length > 2) {
    console.log(res);
  }
});

import { fromEvent } from 'rxjs';
import { filter, map } from 'rxjs/operators';

/**
 * rxjs 实现
 */

const input$ = fromEvent(document.querySelector('.input1'), 'input');

input$
  .pipe(
    filter(
      (event: InputEvent) => (event.target as HTMLInputElement).value.length > 2
    ),
    map((event: InputEvent) => (event.target as HTMLInputElement).value)
  )
  .subscribe((value: string) => console.log(value));

看起来没有太多优势? 从代码量上来讲，还有可能劣化了？

问2: 上面的输入框每次输入之后都动态保存到后端，如何实现?

/**
 * 常规命令式实现
 */

// 防抖 自己实现 or lodash 之类的方法库
function debounce(fn, delay = 500) {
  let timer = null;

  return function () {
    if (timer) {
      clearTimeout(timer);
    }
    timer = setTimeout(() => {
      fn.apply(this, arguments);
      timer = null;
    }, delay);
  };
}

const input2$ = document.querySelector('.input2');
// 同步
input2$.addEventListener('input', debounce(async (event: Event) => {
    const res = (event.target as HTMLInputElement).value;
    if (res.length > 2) {
      // 异步, 保存数据
      await saveData();
    }
}, 600));

import { fromEvent } from 'rxjs';
import { filter, map, debounceTime, tap } from 'rxjs/operators';

/**
 * rxjs 实现
 */

const input$ = fromEvent(document.querySelector('.input1'), 'input');

input$
  .pipe(
    filter(
      (event: InputEvent) => (event.target as HTMLInputElement).value.length > 2
    ),
    map((event: InputEvent) => (event.target as HTMLInputElement).value),
    debounceTime(600),
    tap(saveData)
  )
  .subscribe();

在线代码: https://stackblitz.com/edit/rxjs-u1sphw?file=index.ts

🙋 解决什么问题?

使用响应式编程来解决同步/异步混用问题。

最大目的是提供一系列抽象的操作符可以对数据进行转换，而不管这些数据来源是同步或异步的。

输入 - 所有行为转换为流
输出 - 统一使用副作用
流转 - 操作符优雅的时序控制

理解点1：流(streams)

流(streams): 随时间流逝的一系列事件。

举个例子:

image.png

流的概念具体在 rxjs 里面，就是 Observerable 和 Observer 的关系, 如下图所示

它有发送数据的能力（Observerable）
它有接受数据的能力 (Observer,Subscribe)
它能对数据进行转换 (Operator)

image.png

// 创建一个流
// RxJS v6+
import { Observable } from 'rxjs';
/*
  创建在订阅函数中发出 'Hello' 和 'World' 的 observable 。
*/
const hello = new Observable(function(observer) {
  observer.next('Hello');
  observer.next('World');
});

// 输出: 'Hello'...'World'
const subscribe = hello.subscribe(val => console.log(val));

https://stackblitz.com/edit/typescript-baxh98?file=index.ts&devtoolsheight=100

理解点2: Observable & observer & Subscription

定义

Observable (可观察对象): 表示一个一个可调用的未来值或事件的集合。
Observer (观察者): 一个回调函数的集合，它知道如何去监听由 Observable 提供的值。
Subscription (订阅): 表示 Observable 的执行，主要用于取消 Observable 的执行。

import { Observable } from 'rxjs';

const node = document.querySelector('input');
// input$ 可观察对象 Observable
const input$ = Observable.fromEvent(node, 'input');

/**
* 以下的对象为完整的observer
*   {
    next: (event) => console.log(`You just typed ${event.target.value}!`),
    error: (err) => console.log(`Oops... ${err}`),
    complete: () => console.log(`Complete!`)
    }
*  也可以使用 .subscribe(console) 来简写next的输入  
*/
const subscription = input$.subscribe({
  next: (event) => console.log(`You just typed ${event.target.value}!`),
  error: (err) => console.log(`Oops... ${err}`),
  complete: () => console.log(`Complete!`)
});

// 取消监听
subscription.unsubscribe();

与Promise对比

由于 Observable 的创建方式和 Promise 有点像，因为用他们做对比学习可能效果更好。

创建流程
promise创建流程

image.png

promise 是一次性的，在异步任务执行完毕后，promise 就被标记为 fullfilled 或者 rejected 状态。

Observable创建流程

Observable 不是一次性的，在异步任务中可以通过 next 多次触发。只有收到 error 或者 complete，订阅才会结束。

惰性定义

当创建事件发送的逻辑时，所有的逻辑是在订阅后才执行。

import { Observable } from "rxjs";

//创建 promise
const promise = new Promise(resolve => {
  console.log("run promise");
  setTimeout(() => resolve("ok"), 2000);
});

//订阅promise
// promise.then(console.log);

//创建 observable
const observable = new Observable(subscriber => {
  console.log("run observable");
  setTimeout(() => {
    subscriber.next("ok");
    subscriber.complete();
  }, 2000);
});

// observable.subscribe(console.log);

多值

promise是只为单个值设计的，所以当收到一个值之后这个 promise 就结束了。但是 observable不是，只要没有 complete 或者出现 error，它可以发送多个值。

import { Observable } from "rxjs";

//创建 observable
const observable = new Observable(subscriber => {
  let count = 0;
  const id = setInterval(() => {
    subscriber.next(++count);
  }, 1000);

  return () => {
    clearInterval(id);
  };
});

//每间一秒加一
const sub = observable.subscribe(console.log);
//两秒后取消监听，导致事件停止发送
setTimeout(() => sub.unsubscribe(), 2000);

Promise 和 Observable的对比