RXJS如何向已有Observable流中动态添加新Observable？

现有如下代码：

class X {
  subjects: Subject<any>[] = [new Subject<any>()];
  
  listen(): Observable<any> {
    return merge(...this.subjects).pipe();
  }
}

class Y {
   constructor() {
     new X().listen().subscribe(value => console.log('received', value));
   }
}

给X添加addSubject方法后：

class X {
  subjects: Subject<any>[] = [new Subject<any>()];
  
  listen(): Observable<any> {
    return merge(...this.subjects).pipe();
  }

  addSubject(subjectToAdd: Subject<any>): void {
    this.subjects.push(subjectToAdd);
  }
}

此时在新增的subjectToAdd上发送事件，Y完全收不到——因为merge(...this.subjects)只会合并调用listen时存在的Subject，后续新增的不会被纳入合并范围。

需要实现的效果：Y只需要订阅X的listen方法，不用管X内部的状态，就能收到X里所有当前已存在和后续添加的Subject的事件。

方案一：用主Subject中转所有事件 #

这是最直接的实现方式，通过一个主Subject来转发所有Subject的事件，不管是初始的还是后续添加的。

修改后的X类：

import { Subject, Observable } from 'rxjs';

class X {
  private mainSubject = new Subject<any>();
  private subjects: Subject<any>[] = [];

  constructor() {
    // 初始化第一个Subject
    const initialSubject = new Subject<any>();
    this.addSubject(initialSubject);
  }

  listen(): Observable<any> {
    return this.mainSubject.asObservable();
  }

  addSubject(subjectToAdd: Subject<any>): void {
    this.subjects.push(subjectToAdd);
    // 把新Subject的事件转发到主Subject
    subjectToAdd.subscribe(this.mainSubject);
  }
}

测试代码（Y类）：

class Y {
  constructor() {
    const x = new X();
    x.listen().subscribe(val => console.log('received:', val));

    // 新增一个Subject并发送事件
    const newSub = new Subject<any>();
    x.addSubject(newSub);
    newSub.next('Hello from new subject!'); // 这里Y会收到这条消息
  }
}

方案二：用BehaviorSubject动态跟踪Subject列表 #

如果想更贴合RxJS的响应式风格，可以用BehaviorSubject存储Subject列表，每次列表变化时重新合并所有Subject：

import { Subject, Observable, merge, BehaviorSubject } from 'rxjs';
import { switchMap } from 'rxjs/operators';

class X {
  private subjects$ = new BehaviorSubject<Subject<any>[]>([]);

  constructor() {
    this.addSubject(new Subject<any>());
  }

  listen(): Observable<any> {
    return this.subjects$.pipe(
      // 每次Subject列表更新，重新合并所有Subject
      switchMap(subjects => merge(...subjects))
    );
  }

  addSubject(subjectToAdd: Subject<any>): void {
    this.subjects$.next([...this.subjects$.value, subjectToAdd]);
  }
}

原代码问题分析 #

原来的merge(...this.subjects)在调用listen时就已经固定了要合并的Observable集合，后续往subjects数组里加新的Subject，不会影响已经创建好的merge Observable。

方案优势 #

• 方案一性能更好，无需每次添加Subject都重新创建merge Observable，仅做简单的事件转发。
• 方案二更符合响应式编程思想，通过状态变化驱动数据流自动更新。

两种方案都能满足需求：Y只需要调用listen订阅，完全不用关心X内部的Subject管理逻辑，就能接收所有事件。

内容的提问来源于stack exchange，提问作者Wouter Vandenputte