[译] 认识 rxjs 中的 BehaviorSubject、ReplaySubject 以及 AsyncSubject

原文连接：Understanding rxjs BehaviorSubject, ReplaySubject and AsyncSubject
原文做者：Luuk Gruijs；发表于2018年5月4日
译者：yk；如需转载，请注明出处，谢谢合做！

Subject 的做用是实现 Observable 的多播。因为其 Observable execution 是在多个订阅者之间共享的，因此它能够确保每一个订阅者接收到的数据绝对相等。不只使用 Subject 能够实现多播，RxJS 还提供了一些 Subject 的变体以应对不一样场景，那就是：BehaviorSubject、ReplaySubject 以及 AsyncSubject。

若是你还不知道 Subject（主题）是什么的话，建议先读读个人上一篇文章：认识 rxjs 中的 Subject。若是你以为 OK 没问题，那咱继续！

摄影：Cory Schadt，来自 Unsplash

BehaviorSubject

BehaviorSubject 是 Subject 的变体之一。BehaviorSubject 的特性就是它会存储“当前”的值。这意味着你始终能够直接拿到 BehaviorSubject 最后一次发出的值。

有两种方法能够拿到 BehaviorSubject “当前”的值：访问其 .value 属性或者直接订阅。若是你选择了订阅，那么 BehaviorSubject 将直接给订阅者发送当前存储的值，不管这个值有多么“久远”。请看下面的例子：

import * as Rx from "rxjs";

const subject = new Rx.BehaviorSubject(Math.random());

// 订阅者 A
subject.subscribe((data) => {
  console.log('Subscriber A:', data);
});

subject.next(Math.random());

// 订阅者 B
subject.subscribe((data) => {
  console.log('Subscriber B:', data);
});

subject.next(Math.random());

console.log(subject.value)

// 输出
// Subscriber A: 0.24957144215097515
// Subscriber A: 0.8751123892486292
// Subscriber B: 0.8751123892486292
// Subscriber A: 0.1901322109907977
// Subscriber B: 0.1901322109907977
// 0.1901322109907977
复制代码

详细讲解一下：

1. 咱们首先建立了一个 subject，并在其中注册了一个订阅者A。因为 subject 是一个 BehaviorSubject，因此订阅者A 会当即收到 subject 的初始值（一个随机数），并同时将其打印。
2. subject 随即广播下一个值。订阅者A 再次打印收到的值。
3. 第二次订阅 subject 并称其为订阅者B。一样，订阅者B 也会当即收到 subject 当前存储的值并打印。
4. subject 再次广播下一个新的值。这时，两个订阅者都会收到这个值并打印。
5. 最后，咱们经过简单地访问 .value 属性的形式获取了 subject 当前的值并打印。这在同步的场景下很是好用，由于你没必要订阅 Subject 就能够获取它的值。

另外，你可能发现了 BehaviorSubject 在建立时是须要设置一个初始值的。这一点在 Observable 上就很是难实现，而在 BehaviorSubject 上，只要传递一个值就好了。

译者注：当前版本的 RxJS 中 BehaviorSubject() 是必需要设置初始值的，不然会致使执行错误，而原文并无体现这一点。因此我在这一段作了较多改动，以避免误导读者。详见 BehaviorSubject。

ReplaySubject

相比 BehaviorSubject 而言，ReplaySubject 是能够给新订阅者发送“旧”数据的。另外，ReplaySubject 还有一个额外的特性就是它能够记录一部分的 observable execution，从而存储一些旧的数据用来“重播”给新来的订阅者。

当建立 ReplaySubject 时，你能够指定存储的数据量以及数据的过时时间。也就是说，你能够实现：给新来的订阅者“重播”订阅前一秒内的最后五个已广播的值。示例代码以下：

import * as Rx from "rxjs";

const subject = new Rx.ReplaySubject(2);

// 订阅者 A
subject.subscribe((data) => {
  console.log('Subscriber A:', data);
});

subject.next(Math.random())
subject.next(Math.random())
subject.next(Math.random())

// 订阅者 B
subject.subscribe((data) => {
  console.log('Subscriber B:', data);
});

subject.next(Math.random());

// Subscriber A: 0.3541746356538569
// Subscriber A: 0.12137498878080955
// Subscriber A: 0.531935186034298
// Subscriber B: 0.12137498878080955
// Subscriber B: 0.531935186034298
// Subscriber A: 0.6664809293975393
// Subscriber B: 0.6664809293975393
复制代码

简单解读一下代码：

1. 咱们建立了一个 ReplaySubject 并指定其只存储最近两次广播的值；
2. 订阅 subject 并称其为订阅者A；
3. subject 连续广播三次，同时订阅者A 也会跟着连续打印三次；
4. 这一步就轮到 ReplaySubject 展示魔力了。咱们再次订阅 subject 并称其为订阅者B，由于以前咱们指定 subject 存储最近两次广播的值，因此 subject 会将上两个值“重播”给订阅者B。咱们能够看到订阅者B 随即打印了这两个值；
5. subject 最后一次广播，两个订阅者收到值并打印。

以前提到了你还能够设置 ReplaySubject 的数据过时时间。让咱们来看看下面这个例子：

import * as Rx from "rxjs";

const subject = new Rx.ReplaySubject(2, 100);

// 订阅者A
subject.subscribe((data) => {
  console.log('Subscriber A:', data);
});

setInterval(() => subject.next(Math.random()), 200);

// 订阅者B
setTimeout(() => {
  subject.subscribe((data) => {
    console.log('Subscriber B:', data);
  });
}, 1000)

// Subscriber A: 0.44524184251927656
// Subscriber A: 0.5802631630066313
// Subscriber A: 0.9792165506699135
// Subscriber A: 0.3239616040117268
// Subscriber A: 0.6845077617520203
// Subscriber B: 0.6845077617520203
// Subscriber A: 0.41269171141525707
// Subscriber B: 0.41269171141525707
// Subscriber A: 0.8211466186035139
// Subscriber B: 0.8211466186035139
复制代码

一样解读一下代码：

1. 咱们建立了一个 ReplaySubject，指定其存储最近两次广播的值，但只保留 100ms；
2. 订阅 subject 并称其为订阅者A；
3. 咱们让这个 subject 每 200ms 广播一次。订阅者A 每次都会收到值并打印；
4. 咱们设置在程序执行一秒后再次订阅 subject，并称其为订阅者B。这意味着在它开始订阅以前，subject 就已经广播过五次了。因为咱们在建立 subject 时就设置了数据的过时时间为 100ms，而广播间隔为 200ms，因此订阅者B 在开始订阅后只会收到前五次广播中最后一次的值。

AsyncSubject

BehaviorSubject 和 ReplaySubject 均可以用来存储一些数据，而 AsyncSubject 就不同了。AsyncSubject 只会在 Observable execution 完成后，将其最终值发给订阅者。请看代码：

import * as Rx from "rxjs";

const subject = new Rx.AsyncSubject();

// 订阅者A
subject.subscribe((data) => {
  console.log('Subscriber A:', data);
});

subject.next(Math.random())
subject.next(Math.random())
subject.next(Math.random())

// 订阅者B
subject.subscribe((data) => {
  console.log('Subscriber B:', data);
});

subject.next(Math.random());
subject.complete();

// Subscriber A: 0.4447275989704571
// Subscriber B: 0.4447275989704571
复制代码

虽然这段代码的输出并很少，但咱们仍是照例解读一下：

1. 建立 AsyncSubject；
2. 订阅 subject 并称其为订阅者A；
3. subject 连续广播三次，但什么也没发生；
4. 再次订阅 subject 并称其为订阅者B；
5. subject 又广播了一次，但仍是什么也没发生；
6. subject 完成。两个订阅者这才收到传来的值并打印至终端。

结论

BehaviorSubject、ReplaySubject 和 AsyncSubject 均可以像 Subject 同样被用于多播。它们各自都有一些额外的特性以适用于不一样场景。

译者注：这个结论好敷衍。。。