rxjs 基础使用

Posted 2022-12-21 GoldenaArcher

rxjs 基础使用

rxjs 主要就是一个异步的解决方案，目前项目在用，也就学学看，想看看能不能用 rxjs 解决现在业务上的一个需求。

rxjs 官方都有说：

Think of RxJS as Lodash for events.

想一下吧 RxJS 当作处理事件的 Lodash

基本上就可以说它封装了很多处理事件的工具用于简化操作流程。

rxjs 的一个简单的使用案例：

这个是 util 的类：

import  Observable, of  from 'rxjs';

export const name$ = of('Alice', 'Ben', 'Charlie');

export function storeDataOnServer(data: string): Observable<string> 
  return new Observable((subscriber) => 
    setTimeout(() => 
      subscriber.next(`Saved successfully: $data`);
      subscriber.complete();
    , 5000);
  );


export function storeDataOnServerError(data: string): Observable<string> 
  return new Observable((subscriber) => 
    setTimeout(() => 
      subscriber.error(new Error('Failure!'));
    , 5000);
  );

这个是 index.js：

import  name$, storeDataOnServer, storeDataOnServerError  from './external';

name$.subscribe((value) => console.log(value));

storeDataOnServer('some value').subscribe((val) => console.log(val));

storeDataOnServerError('failed value').subscribe(
  next: (val) => console.log(val),
  error: (err) => console.error(err.message),
);

输出的结果：

名词

Stream

Stream 的定义如下：

Potentially infinite analog of a list

一个可能无限传输信号的列表

本质上来说，rxjs 操作的就是流。

Observable

rxjs 对于 Observable 的解释如下：

我的理解就是可以把 Observable 视作 Promise 的一个超集（像 JS 和 TS 之间的关系），只不过 Promise 处理一个事件，而 Observable 可以处理多个事件。

官方案例如下：

content_copyopen_in_new;
import  Observable  from 'rxjs';

const observable = new Observable((subscriber) => 
  subscriber.next(1);
  subscriber.next(2);
  subscriber.next(3);
  setTimeout(() => 
    subscriber.next(4);
    subscriber.complete();
  , 1000);
);

⚠️：如果只创建一个 Observable，那么什么都不会发生，就像装薛定谔猫的那个盒子，没有打开，它就永远只装着薛定谔的猫，你也不知道里面的状态到底是什么，也不知道里面的结果是什么。

Subscription

A Subscription is an object that represents a disposable resource, usually the execution of an Observable. A Subscription has one important method, unsubscribe, that takes no argument and just disposes the resource held by the subscription. In previous versions of RxJS, Subscription was called “Disposable”.

官方文档的解释如上，提炼的重点有几个：

• Subscription 是一个对象

• Subscription 是一个可抛弃的资源

• Subscription 是 Observable 的一个执行

• Subscription 有一个重要的函数是 unsubscribe

  调用了 unsubscribe 就相当于废弃了这个资源（提前终止），这也就是为什么说 Subscription 是一个可抛弃的资源

这是 Subscription 的官方案例，其实比起上面的 Observable 来说只是多了一个 unsubscribe 的调用。

import  interval  from 'rxjs';

const observable = interval(1000);
const subscription = observable.subscribe((x) => console.log(x));
// Later:
// This cancels the ongoing Observable execution which
// was started by calling subscribe with an Observer.
subscription.unsubscribe();

个人理解就是，subscription 就是一个执行，依旧用薛定谔的猫做案例，subscription 这个步骤就是去打开那个盒子，但是在这个过程中（盒子还是关闭的状态下），依旧不会知道结果。

⚠️：如果提前终止了会触发 side effect 的操作而没有清理这个 side effect（比如说定时器等），这就可能会出现内存泄露，如：

这个案例中只是一个简单的 log，但是如果第二个 timeout 里面会有其他的操作，那么就到时候虽然 rxjs 没有执行，但是其他地方却执行了这个代码，从而导致不可知的后果。

⚠️:每一个 subscription 都会创建一个新的 instance：

可以看到这个案例中，observable 执行了两次（创建两个 instance），并且每一个执行都可以操作原本代码块的变量（这也就是为什么不清理可能会出现内存泄露）。

Observer

An Observer is a consumer of values delivered by an Observable. Observers are simply a set of callbacks, one for each type of notification delivered by the Observable: next, error, and complete. The following is an example of a typical Observer object

Observer 就是最后消耗结果的那部分，也就是说执行打开箱子这个操作完成后，应该拿那只猫怎么办。

它里面包含了三个回调函数：

• next

  既成功后进行的操作

• error

  失败后的操作

  失败后会终止整个 Observable

• complete

  完成后的操作

  完成后也会终止整个 Observable

官方提供的案例如下：

content_copyopen_in_new;
const observer = 
  next: (x) => console.log('Observer got a next value: ' + x),
  error: (err) => console.error('Observer got an error: ' + err),
  complete: () => console.log('Observer got a complete notification'),
;

可以看到，complete 是没有参数的。

到这一步，rxjs 基础使用就已经完成了，比如说可以将 http 调用放到 observable 中，随后创建 observer 对获取的结果进行操作。