作为前端，你需要知道 RxJS（响应式编程-流）

RxJS.png

前言

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

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

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

什么是 RxJS

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

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

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

  • 控制异步任务的执行顺序。

  • 控制异步任务的开始和暂停。

  • 可以获取异步任务执行的各个阶段的状态。

  • 监听异步任务执行过程中发生的错误，在并发执行的情况下，不会因为一个异步任务出错而影响到其他异步代码的执行。

RxJS 的真实面目（流）

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

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

Node 中的 Stream

「Node」 中的 Stream 指的是一个「抽象接口」，例如文件读取、「Http Request」都实现了这个接口。并且，每一个「流」都是 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」中是以创建任务的形式使用「流」。

例如，我们想要配置一个 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 dafult 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()、map、flatMap等等。

  1. Subject 主体

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

例如：

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() 中的代码块放在消息队列中以宏任务 setTimeout 或 setInterval 的形式执行，默认时钟设置为 0。

小结

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

  • 控制流

  • 进行全局的状态存储

  • 结合第三方状态存储 immutable 使用

  • 结合 React 使用

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

写在最后

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

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

以上是 作为前端，你需要知道 RxJS（响应式编程-流） 的全部内容， 来源链接： utcz.com/a/27351.html

回到顶部