Rxjs和它的一些概念 | ChangerHe’s Blog

password

Created

Jun 2, 2023 08:13 AM

type

Post

status

Published

date

Jun 2, 2023

slug

summary

Rxjs是什么

在这之前我们需要理解一个概念是”纯函数”与”副作用函数”

纯函数指不会产生副作用的函数, 也就是说, 其输出完全由输入决定, 无论调用多少次, 输出特定值一定得到的是相应的结果

副作用函数, 指调用函数时, 除了返回函数值外, 还会对主调用函数产生附加的影响, 例如修改全局变量, 修改参数或改变外部存储等

通常, 如果我们创建一个非纯函数, 在这个函数之外使用了共享变量的代码, 会使应用状态混乱难以维护

这里Rxjs给出的方案则是: 响应式编程

响应式编程或反应式编程（Reactive programming）是一种面向数据流和变化传播的声明式编程范式。这意味着可以在编程语言中很方便地表达静态或动态的数据流，而相关的计算模型会自动将变化的值通过数据流进行传播。

用通俗的话讲就是, Rxjs会将数据从源头进行包装, 包装后的数据可以进行观测, 从而保证了数据源头变化的时候, 下游观测的数据跟着一起变化, 并传播下去.

对于开发者来讲, 就只需要先定义好数据源头的处理机制, 及其观测数据的响应机制即可

什么时候使用Rxjs

关于Rxjs, 这里引用一个比较好的回答

思考一下，异步的本质是什么？
异步操作和同步操作最大的区别就是异步有时序。
我们可以把同步操作理解为：数据+函数
那么异步操作就是：数据+函数+时序
Rx就是把时序抽离成一根时间轴，在这根时间轴上进行同步操作，而异步相关的时序处理就交给Rx提供的各种operator操作符。
所以问题就很简单了，如果你的应用是一个时序密集的应用，那么使用Rx能帮你理清复杂的异步逻辑。反之，如果异步操作之间没有太多的联系，时序分散， 则不那么需要使用Rx。

Rxjs的使用

基本语法

进阶使用

Rxjs基于其设计理念, 也对应的会有一些相应的抽象概念, 主要是下面几个

Observable 可观测对象

将一个数据流看作一个可观察对象，表示这个数据流变化传播过程中发生的一些列事件的集合。

Observable可以随着时间推移返回多个值


Observable
const foo = Observable.create(function (observer) {
    observer.next('Hello');
    observer.next('world');
});

foo.subscribe(function (x) {
    console.log(x);
});

// 控制台输出
'Hello'
'world'

// 也可以异步推送一些值
const foo = Observable.create(function (observer) {
    observer.next('Hello')；
    setTimeout(() => {
        observer.next('rxjs');
    },0)
    observer.next('world');
});

// 控制台输出
'Hello'
'world'
'rxjs'

observable一般会有三种操作

next 推送一个值, 代表”下一步的结果”的意思

error: 推送一个异常, 代表流执行到这里的时候终止了

complete: 推送一个已完成的消息, 表示不会再发送任何一个值了


Observable.create(observer => {
    try {
        observer.next(1);
        observer.next(2);
        observer.complete();
        observer.next(3); // 前面已经通知观察者已经完成，所以这个值不会发送
    } catch (e) {
        observer.error(e); // 捕获到异常发送一个错误
    }
})

同时, 在整个流程结束之后, 我们可以将订阅关系取消掉


// 调用subscribe时，观察者会被附加到新创建的Observable执行中，
// 会返回一个对象，即Subscription（订阅）
var subscription = observable.subscribe();
// Subscription表示正在进行中的执行，调用unsubscribe()来取消observable执行；
subscription.unsubscribe();

Observer 观察者

Observer（观察者）是一组回调函数的集合，每一个回调函数对应Observable发送通知的类型：next、error、complete。


const observer = {
    next: () => {}, // 观察者接收到next()消息执行的回调函数
    error: () => {}, // 观察者接收到error()消息执行的回调函数
    complete: () => {}, // 接收到complete()消息执行的回调函数
}
// observer中的观察者可能是部分的，没有提供某个回调，observable还是可以执行的。
// 方法1：将observer观察者传入subscribe
observable.subscribe(observer)
// 方法2：subscribe按顺序（next，error，complete)传入三个回调函数
observable.subscribe((value) => {},(error) => {}, () => {})

Subscription 订阅

是一个可清理资源的对象，代表Observable的执行

Subject 主体

主要是针对多个观察者的情况下

如果两个观察者同时观察了某一个observable对象, 但是A观察者的订阅时间早于B, 那么在期间发生的数据变化, B应不应该收到呢?

这里则是有两种做法:

期间的变化数据因为B还没开始订阅, 就不处理了这称为 Hot Observable

B订阅之后, 会从头开始处理变化的数据, 这称为 Cold Observable

对于多个订阅的情况, Rxjs定义了Subject这个概念, 目前有四种不同类型的Subject

Subject: 将值多播给已订阅的该Subject的观察者列表

BehaviorSubject: 把最后一个值（当前值）发送给观察者（需要一个初始值）

AsyncSubject: 执行的最后一个值发给观察者

ReplaySubject: 可以把之前错过的值发给观察者

BehaviorSubject

保存发送给观察者的最后一个值, 如果有新的观察者是, 会立即收到这个最后一个值


var subject = new BehaviorSubject(0); // 0是初始值

subject.subscribe({
  next: (v) => console.log('observerA: ' + v)
});

subject.next(1);
subject.next(2);

subject.subscribe({
  next: (v) => console.log('observerB: ' + v)
});

subject.next(3);
// 输出：
observerA: 0 //line3 ：A订阅时立即收到当前值（初始值）0
observerA: 1 //line7 : BS推送新的值1，订阅者A接收到值1
observerA: 2 //line8 : BS推送新的值2，订阅者A接收到值2
observerB: 2 //line 10 : B订阅时立即收到变化后的当前值2
observerA: 3 //line 14： BS推送新的值3，订阅者A和B一起收到值3
observerB: 3

AsyncSubject

只有当Observable执行完成时【执行complete()】，才会将执行的最后一个值发送给观察者


var subject = new Rx.AsyncSubject();

subject.subscribe({
  next: (v) => console.log('observerA: ' + v)
});

subject.next(1);
subject.next(2);
subject.next(3);
subject.next(4);

subject.subscribe({
  next: (v) => console.log('observerB: ' + v)
});

subject.next(5);
subject.complete();
// 输出：
// line 17 执行complete()后两个订阅者A和B才会收到最后的一个值(5)
observerA: 5
observerB: 5

ReplaySubject

可以发送旧值给新的观察者，还可以记录Observable的执行的一部分，将Observable执行过程中的多个值回放给新的观察者。


var subject = new ReplaySubject(3); // 为新的订阅者缓冲3个值

subject.subscribe({
  next: (v) => console.log('observerA: ' + v)
});

subject.next(1);
subject.next(2);
subject.next(3);
subject.next(4);

subject.subscribe({
  next: (v) => console.log('observerB: ' + v)
});

subject.next(5);
// 输出
observerA: 1 // line 7: RS推送值1，订阅者A收到值1
observerA: 2 // line 8: RS推送值2，订阅者A收到值2
observerA: 3 // Line 9: RS推送值3，订阅者A收到值3
observerA: 4 // line 10: RS推送值4，订阅者A收到值4
observerB: 2 // line 12: 新的订阅者订阅RS
observerB: 3 // 订阅时按顺序收到了RS缓冲的三个值
observerB: 4
observerA: 5 // line 16：RS推送值5，观察者A和B收到值5
observerB: 5

Tips
//RS除了可以指定缓冲数量，还可以指定时间（单位毫秒）来确定多久之前的值要记录
var subject = new ReplaySubject(3,500) // 记录3个值，500ms前。