RxJS 主题

什么是主题？RxJS 主题是一种特殊的 Observable 类型，它允许将值多播到许多 Observer。普通的 Observable 是单播的（每个订阅的 Observer 拥有 Observable 的独立执行），而 Subject 是多播的。

主题就像一个可观察对象，但是可以多播到许多观察者。主题就像 EventEmitters：它们维护着许多侦听器的注册表。

每个主题都是可观察的。给定一个主题，您可以 subscribe 提供一个观察者，该观察者将开始正常接收值。从观察者的角度来看，它无法确定观察到的执行是来自纯单播观察到的还是主题。

在主题内部，subscribe 不调用传递值的新执行。它将给定的观察者简单地注册到观察者列表中，类似于 addListener 其他库和语言中的正常工作方式。

每个主题都是观察者。它与方法的对象 next(v)，error(e) 和 complete()。要将新值提供给主题，只需调用 next(theValue)，它将被多播到注册以监听主题的观察者。

在下面的示例中，我们将两个观察者附加到一个主题，并将一些值提供给该主题：

import { Subject } from 'rxjs';


const subject = new Subject<number>();


subject.subscribe({
  next: (v) => console.log(`observerA: ${v}`)
});
subject.subscribe({
  next: (v) => console.log(`observerB: ${v}`)
});


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


// Logs:
// observerA: 1
// observerB: 1
// observerA: 2
// observerB: 2

由于主题是观察者，因此这也意味着您可以将主题作为 subscribe 任何可观察对象的参数提供，如以下示例所示：

import { Subject, from } from 'rxjs';


const subject = new Subject<number>();


subject.subscribe({
  next: (v) => console.log(`observerA: ${v}`)
});
subject.subscribe({
  next: (v) => console.log(`observerB: ${v}`)
});


const observable = from([1, 2, 3]);


observable.subscribe(subject); // You can subscribe providing a Subject


// Logs:
// observerA: 1
// observerB: 1
// observerA: 2
// observerB: 2
// observerA: 3
// observerB: 3

通过上面的方法，我们基本上只是通过 Subject 将单播 Observable 执行转换为多播。这说明了主题是如何将任何可观察的执行共享给多个观察者的唯一方法。

还有的几个特例 Subject 类型：BehaviorSubject，ReplaySubject，和AsyncSubject。

组播可观察物

“多播可观察”通过可能有许多订户的主题传递通知，而普通的“单播可观察”仅将通知发送给单个观察者。

多播的 Observable在幕后使用一个 Subject来使多个 Observer看到相同的 Observable 执行。

在幕后，这是 multicast 操作员的工作方式：观察者订阅了基础主题，而主题订阅了源 Observable。以下示例与之前的示例相似 observable.subscribe(subject)：

import { from, Subject } from 'rxjs';
import { multicast } from 'rxjs/operators';


const source = from([1, 2, 3]);
const subject = new Subject();
const multicasted = source.pipe(multicast(subject));


// These are, under the hood, `subject.subscribe({...})`:
multicasted.subscribe({
  next: (v) => console.log(`observerA: ${v}`)
});
multicasted.subscribe({
  next: (v) => console.log(`observerB: ${v}`)
});


// This is, under the hood, `source.subscribe(subject)`:
multicasted.connect();

multicast 返回一个看起来像普通 Observable 的 Observable，但是在订阅时却像 Subject 一样工作。 multicast返回一个 ConnectableObservable，它只是该 connect()方法的 Observable 。

该 connect()方法对于准确确定共享 Observable 执行何时开始非常重要。因为 connect()确实 source.subscribe(subject)引擎盖下，connect()返回订阅，这可以从以取消共享可观测执行取消订阅。

参考计数

connect() 手动调用和处理订阅通常很麻烦。通常，我们要在第一个观察者到达时自动连接，并在最后一个观察者取消订阅时自动取消共享执行。

请考虑以下示例，在此示例中，如列表所概述的那样发生订阅：

1. 第一观察员订阅多播的 Observable
2. 多播的 Observable 已连接
3. 该 next 值 0 传递给第一位观察者
4. 第二观察者订阅多播的 Observable
5. 该 next 值 1 传递给第一位观察者
6. 该 next 值 1 传递给第二个观察者
7. 第一观察员从多播的 Observable 退订
8. 该 next值 2 传递给第二个观察者
9. 第二观察者从多播的 Observable 退订
10. 与多播 Observable 的连接已取消订阅

为了实现对的显式调用 connect()，我们编写了以下代码：

import { interval, Subject } from 'rxjs';
import { multicast } from 'rxjs/operators';


const source = interval(500);
const subject = new Subject();
const multicasted = source.pipe(multicast(subject));
let subscription1, subscription2, subscriptionConnect;


subscription1 = multicasted.subscribe({
  next: (v) => console.log(`observerA: ${v}`)
});
// We should call `connect()` here, because the first
// subscriber to `multicasted` is interested in consuming values
subscriptionConnect = multicasted.connect();


setTimeout(() => {
  subscription2 = multicasted.subscribe({
    next: (v) => console.log(`observerB: ${v}`)
  });
}, 600);


setTimeout(() => {
  subscription1.unsubscribe();
}, 1200);


// We should unsubscribe the shared Observable execution here,
// because `multicasted` would have no more subscribers after this
setTimeout(() => {
  subscription2.unsubscribe();
  subscriptionConnect.unsubscribe(); // for the shared Observable execution
}, 2000);

如果我们希望避免显式调用 connect()，则可以使用 ConnectableObservable 的 refCount()方法（引用计数），该方法返回一个 Observable，该跟踪可跟踪其拥有的订户数量。当订户数量从 0 增加到时 1，它将要求 connect()我们启动共享执行。只有当订户数量从减少 1 到 0 时，才会完全取消订阅，从而停止进一步执行。

refCount 使多播的 Observable 在第一个订户到达时自动开始执行，并在最后一个订户离开时停止执行。

下面是一个示例：

import { interval, Subject } from 'rxjs';
import { multicast, refCount } from 'rxjs/operators';


const source = interval(500);
const subject = new Subject();
const refCounted = source.pipe(multicast(subject), refCount());
let subscription1, subscription2;


// This calls `connect()`, because
// it is the first subscriber to `refCounted`
console.log('observerA subscribed');
subscription1 = refCounted.subscribe({
  next: (v) => console.log(`observerA: ${v}`)
});


setTimeout(() => {
  console.log('observerB subscribed');
  subscription2 = refCounted.subscribe({
    next: (v) => console.log(`observerB: ${v}`)
  });
}, 600);


setTimeout(() => {
  console.log('observerA unsubscribed');
  subscription1.unsubscribe();
}, 1200);


// This is when the shared Observable execution will stop, because
// `refCounted` would have no more subscribers after this
setTimeout(() => {
  console.log('observerB unsubscribed');
  subscription2.unsubscribe();
}, 2000);


// Logs
// observerA subscribed
// observerA: 0
// observerB subscribed
// observerA: 1
// observerB: 1
// observerA unsubscribed
// observerB: 2
// observerB unsubscribed

该 refCount()方法仅存在于 ConnectableObservable 上，并且返回 Observable，而不是另一个ConnectableObservable。

行为主体

Subject 的变体之一是 BehaviorSubject，其概念为“当前值”。它存储了发给其使用者的最新值，并且每当有新的 Observer 订阅时，它将立即从接收到“当前值” BehaviorSubject。

BehaviorSubjects 对于表示“随时间变化的值”很有用。例如，生日的事件流是主题，而一个人的年龄流将是 BehaviorSubject。

在以下示例中，将使用 0第一个观察者订阅时收到的值初始化 BehaviorSubject 。第二个观察者2即使2在发送值后订阅了该值，也可以接收该值。

import { BehaviorSubject } from 'rxjs';
const subject = new BehaviorSubject(0); // 0 is the initial value


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);


// Logs
// observerA: 0
// observerA: 1
// observerA: 2
// observerB: 2
// observerA: 3
// observerB: 3

重播主题

A ReplaySubject 与 a 相似 BehaviorSubject，它可以将旧值发送给新的订户，但是它也可以记录 Observable 执行的一部分。

A ReplaySubject 记录了来自 Observable 执行的多个值，并将它们重放给新的订户。

创建时 ReplaySubject，您可以指定要重播的值：

import { ReplaySubject } from 'rxjs';
const subject = new ReplaySubject(3); // buffer 3 values for new subscribers


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);


// Logs:
// observerA: 1
// observerA: 2
// observerA: 3
// observerA: 4
// observerB: 2
// observerB: 3
// observerB: 4
// observerA: 5
// observerB: 5

除了缓冲区大小以外，您还可以指定窗口时间（以毫秒为单位），以确定记录的值可以使用多长时间。在下面的示例中，我们使用较大的缓冲区大小 100，但是窗口时间参数仅为 500 毫秒。

import { ReplaySubject } from 'rxjs';
const subject = new ReplaySubject(100, 500 /* windowTime */);


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


let i = 1;
setInterval(() => subject.next(i++), 200);


setTimeout(() => {
  subject.subscribe({
    next: (v) => console.log(`observerB: ${v}`)
  });
}, 1000);


// Logs
// observerA: 1
// observerA: 2
// observerA: 3
// observerA: 4
// observerA: 5
// observerB: 3
// observerB: 4
// observerB: 5
// observerA: 6
// observerB: 6
// ...

异步主题

AsyncSubject 是一个变体，其中只有 Observable 执行的最后一个值发送到其观察者，并且仅在执行完成时发送。

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

AsyncSubject 与 last()运算符类似，因为它等待 complete通知以便传递单个值。