基于 version 6 的 RxJS 入门

分享几个网站
RxJS官网
开发手册

RxJS 是什么?

我们可以简单的认为 RxJS 是一套处理异步编程的 API

RxJS 的特点

• 数据流抽象了很多现实问题
• 擅长处理异步问题
• 把复杂问题分解为简单问题的组合

RxJS 对数据采用 "推" 的方式, 当一个数据产生时, 会将其推送给对应的处理函数, 这个处理函数不需要关心数据是同步产生还是一步产生的, 因此处理异步就会省去很多麻烦, 变得很简单

RxJS 的使用方法

先看一个简单的小例子, 通过这个小例子走进 RxJS 的大门
需求: 点击按钮, 控制台输出 'click'

• 原生JS的实现方式

// index.ts
const button = document.querySelector('button')
button.addEventListener('click', () => console.log('click'))

• RxJS 的实现方式

// index.ts
const button = document.querySelector('button')
const source$ = fromEvent(button, 'click')
source$.subscribe(v => console.log('click'))

演示地址
上面这个例子, 原生 JS 的实现方式大家应该都不陌生, RxJS 中就出现了一些我们不理解的方法, 莫急, 现在不管你理解与否, 我们可以先来了解一下 RxJS 的几个概念

RxJS 一个核心 三个重点

一个核心是 Observable 再加上操作符 Operators
三个重点分别是 Observer、Subject、Schedulers

Observable 是什么

Observable: 可观察的, 这个是百度翻译的解释, 事实上在 RxJS 中, Observable 也有有着差不多的概念 -- 可被观察者
Observer -- 观察者
subscribe -- 订阅
根据字面意思, 我们明显能够感知到 Observable 和 Observer 存在着某种关联, 而在 RxJS 中, 这两者通过 subscribe 方法连在了一起

创建 Observable

要创建一个 Observable, 只要给 new Observable 传递一个接收 Observer 参数的回调函数, 在这个函数中去定义如何发送数据

// index.ts

// 引入 Observable
import { Observable } from 'rxjs'; 
//创建一个 Observable
// 此时的 source$ 就是一个可被观察者
const source$ = new Observable(observer => {
  observer.next(1)
  observer.next(2)
  observer.next(3)
})

// observer 是一个 观察者
const observer = {
  next: item => console.log(item)
}

console.log('start')
source$.subscribe(observer)
console.log('end')
// 输出  start 1 2 3 end

上面这个例子通过 new Observable 创建了一个 Observable, 调用它的 subscribe 方法进行订阅

我们再看一个异步的例子

// index.ts
const source$ = new Observable(observer => {
  let num = 1
  setInterval(() => {
    observer.next(num++)
  }, 1000)
})

const observer = {
  next: item => console.log(item)
}

console.log('start')
source$.subscribe(observer)
console.log('end')
// 输出 start end 1 2 3 4 5 ...

这个例子就是先输出 start, end, 然后每过1秒输出一个递增的数
演示地址

---

通过这个两个例子, 我们至少能明白一点 RxJS 对于同步和异步的行为都是可以处理的

---

观察者 Observer

观察者 Observer 有三个方法的对象:

• next: 当 Observable 发出新的值的时候被调用, 接收这个值作为参数
• complete: 当 Observable 完结, 没有更多的数据时被调用, 值得注意的是, complete 之后的 next 方法是无效的
• error: 当 Observable 内部发生错误时被调用, 之后的 complete next 方法无效

来看几个例子
例一 complete 与 next

// index.ts
const source$ = new Observable(observer => {
  observer.next(1)
  observer.next(2)
  observer.complete()
  observer.next(3)
})

const observer = {
  next: item => console.log(item),
  complete: () => console.log('complete')
}

source$.subscribe(observer)
// 输出 1 2 complete

上面的代码并不会将 3 输出 -- complete 之后的 next 是无效的

例二 complete 与 error

// index.ts
const source$ = new Observable(observer => {  
  try {
    observer.next(1)
    observer.next(2)    
    throw new Error('there is an exception')
    observer.complete()
  } catch (e) {
    observer.error(e)
  }
})

const observer = {
  next: item => console.log(item),
  error: e => console.log(e),
  complete: () => console.log('complete')
}

source$.subscribe(observer)
// 输出 1 2 Error

上面代码并不会将 complete 输出 -- complete 之后的 next 方法是无效的
演示地址

上面代码中的 观察者 是以一个对象的形式创建的, 我们可以直接把函数作为 subscribe 方法的参数

source$.subscribe(
 item => console.log(item),
 e => console.log(e),
 () => console.log('complete')
)
// 参数依次为 next error complete

为了更方便理解, 大家可以参考一下这个图例

数据流

unsubscribe -- 退订

因为 Observable 的执行可能会是无限的, 观察者通常希望在有限的时间内去终止执行, 来避免浪费计算机能力或内存资源

此时 unsubscribe 方法就到了出场的时候

const source$ = new Observable(observer => {  
let num = 1
  setInterval(() => {
    observer.next(num++)
  }, 1000)
})
const observer = {
  next : item => console.log(item)
}
//  subscribe 的调用会返回一个对象 -- subscription,
// subscription 表示进行中的执行, 具体的细节大家可以去官网了解一下
const subscription = source$.subscribe(observer)
setTimeout(() => {
  subscription.unsubscribe() // 取消订阅
}, 5000)

操作符

在 RxJS 中, 操作符是用来处理数据流的, 我们往往需要对数据流做一系列处理, 才交给 Observer, 这时一个操作符就像一个管道(如下图), 数据流进管道, 完成处理, 流出管道!

就像净水器一样, 自来水流过净水器, 经过净水器的过滤处理, 我们可以喝到更纯净的水

我们先来看一个简单的小例子

import { interval } from 'rxjs'; 
import { map } from 'rxjs/operators';

const source$ = interval(1000).pipe(
  map(x => x * x)
)
source$.subscribe(v => console.log('v is '+ v))

这边我们可以看到出现了两个操作符, interval 属于创建类操作符, map 属于转换类操作符;
interval 操作符创建了一个数据流, interval(1000) 会产生一个每隔 1000ms 就会发出一个从0开始递增的数据;
map 操作符和数组的 map 方法类似, 可对数据流进行处理;
上面的 pipe 方法就是数据管道, 会对数据流进行处理, 上面的例子只有一个 map 操作符, 可以添加更多的操作符来对数据进行处理

关于操作符的引用这边简单的说明一下
一般创建类的操作符, 我们引入当时如下:

import { interval, of, from } from 'rxjs'

其他大部分操作符的引入方式如下:

import { map, take, first, last } from 'rxjs/operators'

这边只是简单的介绍了几个, 操作符有很多分类, 每个类目下又有很多具体的操作符!

操作符分类

未完待续...