vlambda博客
学习文章列表

前言

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

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

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

这里可以简单提提「Vue」中依赖收集和派发更新，在 2x 中对应 watcher 和 dep。在 3.0 中对应 effect 和 dep。

那么，回到今天的正题，让我们来领略一下作为「流」中的经典代表之一「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} 'vue-property-decorator'

@Component
export dafult 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等等。

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

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

小结

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

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

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

写在最后

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

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

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