请区分Observables、Subjects、ReplaySubjects与Behaviour Subjects并提供代码示例

Hey there! I get it—RxJS's observable types can feel confusing when you're just reading code snippets without context. Let's break down Observables, Subjects, ReplaySubjects, and BehaviorSubjects clearly, with practical examples that show their key differences.

1. Observable（普通可观察对象） #

核心特性 #

• 属于冷可观察对象：每个订阅者都会触发数据流的独立执行逻辑，相当于给每个订阅者开了一个专属数据流副本。
• 单向数据流：只能由Observable发射数据，订阅者只能接收，无法主动发送数据。

示例代码 #

import { Observable } from 'rxjs';

const observable = new Observable(subscriber => {
  console.log('Observable 开始执行');
  subscriber.next('第一个值');
  setTimeout(() => {
    subscriber.next('延迟1秒的值');
    subscriber.complete();
  }, 1000);
});

// 第一个订阅者
console.log('订阅者1开始订阅');
observable.subscribe({
  next: val => console.log(`订阅者1收到: ${val}`),
  complete: () => console.log('订阅者1完成')
});

// 1.5秒后第二个订阅者加入
setTimeout(() => {
  console.log('订阅者2开始订阅');
  observable.subscribe({
    next: val => console.log(`订阅者2收到: ${val}`),
    complete: () => console.log('订阅者2完成')
  });
}, 1500);

输出解释 #

订阅者1开始订阅
Observable 开始执行
订阅者1收到: 第一个值
订阅者1收到: 延迟1秒的值
订阅者1完成
订阅者2开始订阅
Observable 开始执行
订阅者2收到: 第一个值
订阅者2收到: 延迟1秒的值
订阅者2完成

每个订阅者都会从头触发Observable的执行逻辑，完全独立于其他订阅者。

2. Subject（主题） #

核心特性 #

• 属于热可观察对象：所有订阅者共享同一个数据流，不会重复执行逻辑。
• 既是Observable也是Observer：它可以订阅其他Observable，然后把收到的数据转发给自己的所有订阅者。
• 无历史数据缓存：新订阅者只能收到订阅之后发射的数据，之前的数据会错过。

示例代码 #

import { Subject } from 'rxjs';

const subject = new Subject();

// 第一个订阅者先订阅
subject.subscribe({
  next: val => console.log(`订阅者1收到: ${val}`)
});

// 发送第一个值
subject.next('Hello');

// 第二个订阅者后加入
subject.subscribe({
  next: val => console.log(`订阅者2收到: ${val}`)
});

// 发送第二个值
subject.next('World');

输出解释 #

订阅者1收到: Hello
订阅者1收到: World
订阅者2收到: World

订阅者2在Hello发送后才订阅，所以看不到这个值，只能接收后续的World。

3. ReplaySubject（重放主题） #

核心特性 #

• 热可观察对象，多播共享数据流。
• 缓存历史数据：可以指定缓存最近的N个值，或者指定时间窗口内的历史数据，新订阅者一订阅就会收到这些缓存的历史值，再接收新数据。
• 无需初始值：可以空初始化，后续再发射数据。

示例1：重放最近2个值 #

import { ReplaySubject } from 'rxjs';

const replaySubject = new ReplaySubject(2); // 缓存最近2个值

replaySubject.next('值1');
replaySubject.next('值2');
replaySubject.next('值3');

// 第一个订阅者
replaySubject.subscribe({
  next: val => console.log(`订阅者1收到: ${val}`)
});

// 发送新值
replaySubject.next('值4');

// 第二个订阅者加入
replaySubject.subscribe({
  next: val => console.log(`订阅者2收到: ${val}`)
});

输出解释 #

订阅者1收到: 值2
订阅者1收到: 值3
订阅者1收到: 值4
订阅者2收到: 值3
订阅者2收到: 值4

示例2：按时间窗口重放（最近1秒内的值） #

import { ReplaySubject } from 'rxjs';

// 最多缓存100个值，且只保留最近1秒内的数据
const replaySubject = new ReplaySubject(100, 1000);

replaySubject.next('旧值1');
setTimeout(() => replaySubject.next('旧值2'), 500);
setTimeout(() => replaySubject.next('新值'), 1200);

// 1.5秒后订阅
setTimeout(() => {
  replaySubject.subscribe({
    next: val => console.log(`订阅者收到: ${val}`)
  });
}, 1500);

输出解释 #

订阅者收到: 新值

旧值1和旧值2都超过了1秒的时间窗口，被自动丢弃，新订阅者只能看到新值。

4. BehaviorSubject（行为主题） #

核心特性 #

• 热可观察对象，多播共享数据流。
• 必须初始化默认值：新订阅者会立即收到当前的最新值（包括初始化的默认值），之后接收新数据。
• 相当于ReplaySubject(1)的特殊场景，但强制要求初始值，非常适合表示“当前状态”（比如用户登录状态、主题切换状态）。

示例代码 #

import { BehaviorSubject } from 'rxjs';

// 初始化默认状态为'未登录'
const userStatus$ = new BehaviorSubject<string>('未登录');

// 第一个订阅者获取当前状态
userStatus$.subscribe({
  next: status => console.log(`订阅者1当前状态: ${status}`)
});

// 更新状态
userStatus$.next('已登录');

// 第二个订阅者加入，直接拿到最新状态
userStatus$.subscribe({
  next: status => console.log(`订阅者2当前状态: ${status}`)
});

// 再次更新状态
userStatus$.next('已退出');

输出解释 #

订阅者1当前状态: 未登录
订阅者1当前状态: 已登录
订阅者2当前状态: 已登录
订阅者1当前状态: 已退出
订阅者2当前状态: 已退出

核心差异总结 #

• Observable: 冷流，单播，每个订阅者独立执行数据流，无历史数据重放。
• Subject: 热流，多播，无历史缓存，新订阅者只能收到订阅后的数据，同时可作为Observer订阅其他流。
• ReplaySubject: 热流，多播，可按数量/时间窗口重放历史数据，无需初始值。
• BehaviorSubject: 热流，多播，必须有初始值，新订阅者立即收到当前最新值（永远缓存1个值）。

内容的提问来源于stack exchange，提问作者Kapil Raghuwanshi