作为前端,你需要知道 RxJS(响应式编程-流)

3,996 阅读5分钟

前言

说起「响应式编程」,大家可能并不陌生。但是,直接说「流」这个名称,可能大家会有点愣。其实,「流」的本质和「响应式编程」并不二般,都是衍生于前端经典的设计模式——「观察者模式」。但是,在一定程度上,可以说「流」则是基于这个模式的一个上层抽象,因为它所具备的能力更多、更加强大。

在我的认知里面,我又给「流」划分了一下,「玄学」。

而在我们平常开发中,使用「观察者模式」最经典的就是「Vue」。其中,一方面在「Vue」2x 版本中对应的就是基于 Object.defineProperty() 实现的一套依赖收集派发更新。另一方面,在「Vue」3.0 版本中对应的就是基于 Proxy 实现的一套依赖收集派发更新

这里可以简单提提「Vue」中依赖收集派发更新,在 2x 中对应 watcherdep。在 3.0 中对应 effectdep

那么,回到今天的正题,让我们来领略一下作为「流」中的经典代表之一「RxJS」的魅力。

什么是 RxJS

官方文档介绍:「RxJS」是使用 Observables 的响应式编程的库,它使编写异步或基于回调的代码更容易。

建议,如果不了解 Observable API 的同学,可以移步去学习一番,再来继续阅读本文。

这样一看,我们可能会把「RxJS」来和 Promise 作一些列比较,因为 Promise 也是为了使得编写异步或基于回调的代码更容易而出现的。也确实,很多在谈论「RxJS」的文章都会举例子来比较两者的优缺点。

所以,在这里我就不举例比较,就简单列举几点「RxJS」解决了哪些使用 Promise 存在的痛点,如:

  • 控制异步任务的执行顺序。
  • 控制异步任务的开始和暂停。
  • 可以获取异步任务执行的各个阶段的状态。
  • 监听异步任务执行过程中发生的错误,在并发执行的情况下,不会因为一个异步任务出错而影响到其他异步代码的执行。

RxJS 的真实面目(流)

为什么这里说「RxJS」的真实面目?因为,前面所说的解决异步或基于回调的代码繁琐问题,仅仅是「RxJS」的一个很简单的一面,它真正玄幻的一面,在于它「流」的特性。

相信大家在此之前应该也听过「流」的概念,例如「Node」中的 Stream 流、「gulp」中的 Stream 流。所以,这里我们先来回忆一下这两个经典的用到「流」的知识点。然后,再循序渐进地进入「RxJS」中「流」的世界。

Node 中的 Stream

「Node」 中的 Stream 指的是一个「抽象接口」,例如文件读取、「Http Request」都实现了这个接口。并且,每一个「流」都是 EventEmitter 的实例。

EventEmitter 的本质就是一个简单的「观察者模式」,可以进行事件的分发和监听。

那么,我们通过简单的读取文件的例子,回忆一番「Node」中 Stream 的事件分发和监听:

const fs = require("fs")
let data = ''

// 创建文件读取流
const readFileStream = fs.createReadStream('stream.txt', 'UTF8')

// 监听流数据读取的事件
readFileStream.on(data, (chunk) => {
  data += chunk
})

// 监听流数据读取结束的事件
readFileStream.on(end, (chunk) => {
  console.log(`文件读取的结果是:${data}`)
})

// 监听流读取文件异常的事件
readerStream.on('error', (err) => {
   console.error(err.stack)
});

这里可以看出,文件读取流在没监听 data 的时候,我们并不能拿到需要的文件中的数据。

这个对于所有的「流」都是一样,它需要被订阅才能被使用,因为「流」是惰性的

然后,我们再来回忆一下「Node」中 Stream 的「管道流」的概念,它指的是「流」可以通过 pipe 链式地调用,例如把文件读取流直接写入文件写入流中:

const fs = require("fs")

const readFileStream = fs.createReadStream('readStream.txt', 'UTF8')

const writeFileStream = fs.createWriteStream('outStream.txt', 'UTF8')

readFileStream.pipe(writeFileStream)

gulp 中的 Stream

不知道大家注意到没有,「gulp」官方文档是这样介绍的:gulp.js 基于流(stream)的自动化构建工具。而且,相信用过「gulp」的同学,应该知道在「gulp」中是以创建任务的形式使用「流」。

具体「流」涉及到插件使用、文件处理和监控的细节,有兴趣的同学可以移步gulp官方文档了解。

例如,我们想要配置一个 sass 的编译,它会是这样:

const gulp = require('gulp')
const sass = require('gulp-sass')
// 创建任务,对 scss 目录下的扩展名为 .sccss 执行编译并输出到 css 文件下 
gulp.task('sass-compile', function(){
  return gulp.src('scss/*.scss')
    .pipe(sass({outputStyle: 'expanded'}).on('error', sass.logError))
    .pipe(gulp.dest('css'))
})
// 监听文件变化执行 sass-compile 任务
gulp.watch('scss/*.scss', ['sass-compile'])

可以看出,我们引入的 gulp 模块,本身就是 Stream。所以,也正如「gulp 官方文档」所说的一般:gulp.js 基于流(stream)的自动化构建工具。

那么,回忆了「Node」和「gulp」中的 Stream 之后。接下来,我们就进入「RxJS」中 Stream万千世界

RxJS 中的 Stream

这里,我们通过「RxJS」 中的几个关键字来循序渐进地认知它:

1.Observable 可观察对象

数据就在 Observable 中流动,以及我们可以使用各种「操作符」Operators 来对流进行处理,例如过滤、去重。

例如:

import Rx from 'rxjs/Rx'

const observable = Rx.Observable.from([1,2,3])
observable.filter(val => val >= 2)
          .subscribe(val => console.log(val)) // 2 3

2.Observer 观察者

它是一个「对象集合」,存放用于监听可观察对象的回调。

例如:

import Rx from 'rxjs/Rx'

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

const observer = {
  next: val => console.log(val),
  error: err => console.error(err),
  complete: () => console.log('completed')
}

observable.subscribe(observable)

3.Subscription 订阅

用于取消对可观察对象的订阅,例如在一个组件的生命周期的结束理应取消订阅。

例如:

import Rx from 'rxjs/Rx'
import {Vue, Component} from 'vue-property-decorator'

@Component
export default class Dialog extends Vue {
  private subscription
  
  private created() {
    const observable = Rx.Observable.fromEvent(this.refs.btnCommit, 'click')
    
    this.subscription = observable.subscribe({ next: () => {
      console.log('click')
    }})
  }
  private destoryed() {
    this.subscription.unsubscribe()
  }
}

4.Operators 操作符

它指的是一些数组的工具函数 filter()some()mapflatMap等等。

这里就不举例子了,就和第一个例子使用 filter() 的方式一样。

  1. Subject 主体

它类似于前面提及的「Node」中的 EventEmitter,进行事件的分发,即广播。

需要注意的是,Observable 观察者的事件分发是单播的。

例如:

import Rx from 'rxjs/Rx'

const subject = new Rx.Subject()

subject.subscribe({
  next: val => console.log(`监听对象1:${val}`)
})
subject.subscribe({
  next: val => console.log(`监听对象2:${val}`)
})

subject.next(1)
subject.next(2)
/*
  输出:
  监听对象1:1
  监听对象2:1
  监听对象1:2
  监听对象2:2 
*/

6.Schedules 调度器

它负责对任务的调度,例如控制事件的执行顺序和排序等等。它由这三部分组成:

  • 数据结构,对任务进行顺序存储。
  • 执行上下文,可以表示何时何地执行任务。
  • 一个(虚拟的)时钟,任务的执行会遵循时钟的设置。

例如:

import Rx from 'rxjs/Rx'

const observable = Rx.Observable.create(function (proxyOb) {
  proxyOb.next(1)
  proxyOb.next(2)
  proxyOb.next(3)
  proxyOb.complete()
})
.observeOn(Rx.Scheduler.async)

const observer = {
  next: x => console.log(x),
  error: err => console.error(err),
  complete: () => console.log('completed'),
};

observable.subscribe(finalObserver)

可以看到这里在创建观察者的时候调用了 observerOn(),以及使用了调度器的 async,这里则会按创建时的顺序将所有订阅的 next() 中的代码块放在消息队列中以宏任务 setTimeoutsetInterval 的形式执行,默认时钟设置为 0。

小结

其实,介绍了这几个关键点后,我想大家对于前面提及解决 Promise 的痛点的由来,应该心中明了。并且,针对一些场景 「RxJS」官方文档 也列举了如何使用它,例如:

  • 控制流
  • 进行全局的状态存储
  • 结合第三方状态存储 immutable 使用
  • 结合 React 使用

所以,在这里我就不一一列举了,毕竟文档上已经讲的很好了,照搬讲一遍就没太大必要哈。有兴趣的同学可以自行了解。

写在最后

这篇文章,其实在一个月之前,我就想着总结分享出来,但是各种因素不了了之。这次恰逢「端午节」,专门腾出几个小时时间思考和总结自己这段时间对「RxJS」和「流」的认知。当然,文章中会存在不足的地方,欢迎大家提「Issue」。

并且,个人认为「流」的存在,在将来很可能会改变一些东西。例如,之前携程的一位前辈写的面向Model编程的前端架构方式,其中就提及需要一种脱离平台的响应式编程处理,列举了如「Redux」、「Vue3.0 reactive API」、「RxJS」等等。有兴趣的同学,可以往这方面继续了解。

写作不易,如果你觉得有收获的话,可以帅气三连击!!!