对Object的理解

什么是 Subject？ - RxJS Subject 是一种特殊类型的 Observable，它允许将值多播给多个观察者，所以 Subject 是多播的，而普通的 Observables 是单播的(每个已订阅的观察者都拥有 Observable 的独立执行)。

1. Subjects are multicast(Subject是多播的).
2. Every Subject is an Observable(Subject是Observable).
3. Every Subject is an Observer(Subject是Observer).

Subject内部维护着一个数组, 使用subscribe方法将Observer放入数组, 将要发送的数据循环发送到这个数组里的Observer, 以这种方式实现多播.

• 引入及实例化Subject:
  • import { Subject } from 'rxjs;
  • const subject = new Subject();

生成Object的方式

基础方式

// Subject订阅Observer
subject.subscribe(Observer1);
subject.subscribe(Observer2);
/* ... */
// Observable订阅Subject
observable.subscribe(subject);

使用中间体 multicast

使用multicast中间体装载Subject并返回一个ConnectableObservable, ConnectableObservable是一个特殊的Observable并代表接受订阅, 使用connect方法完成observable与subject的订阅关系

import { multicast } from 'rxjs/operators';
// 生成中间体 multicast
const multicasted = observable.pipe(multicast(subject));
// 订阅Subject
multicasted.subscribe(Observer1);
multicasted.subscribe(Observer2);
/* ... */
// Observable订阅Subject, 内部执行的就是observable.subscribe(subject), 返回一个observable
multicasted.connect();

• 使用refCount生成已经完成connect的Observable, 此时主observable与subject已完成订阅关系, 不需要执行connect方法, refCount取自rxjs/operators, 使用: const refCounted = observable.pipe(multicast(subject), refCount());

Subject的变体

BehaviorSubject

假如第一个Observer已经开始执行订阅推送, 则Subject之后的订阅, 新的Observer接收的订阅推送是从上一个Observer最后订阅推送作为开始

1. 引入: import { BehaviorSubject } from 'rxjs';
2. 实例化, 传入初始值: new BehaviorSubject(val);

ReplaySubject

缓存最后推送的数据数量, 并在新Observer订阅后推送给这个Observer

1. 引入: import { ReplaySubject } from 'rxjs';
2. 实例化, 传入缓存的数量: new ReplaySubject(val);

AsyncSubject

在Subject执行完complete()后向Observer推送最后一次推送的数据, 行为比较抽象, 官网示例:

import { AsyncSubject } from 'rxjs';
const subject = new 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();

// Logs:
// observerA: 5
// observerB: 5