如何基于单播Observable实现多播Subject？自定义实现存疑

嘿，我完全懂你现在的困惑——自己凑出来的Subject实现总感觉哪里不对劲，尤其是捕获Observer的方式确实有点生硬。咱们一步步拆解，看看从单播Observable生成多播Subject的正确思路，顺便搞清楚RxJS的核心逻辑到底是怎么回事。

首先先理清楚核心概念：

• Observable是单播：每一个新的订阅者都会触发一次_subscribe逻辑，相当于给每个订阅者单独开了一条数据流。
• Subject是多播：它本质上既是Observable（可以被订阅），又是Observer（可以接收事件），所有订阅者共享同一条数据流，事件进来时会广播给所有订阅者。
• Angular的EventEmitter就是对Subject的简单包装——把next方法改名为emit而已。

先说说你当前实现的问题 #

你之前的Subject里嵌套了一个Observable，还自己订阅它来触发订阅者遍历，这种方式绕了不必要的弯，而且直接赋值this.observer的逻辑有严重问题：如果这个内部Observable被多次订阅，this.observer会被不断覆盖，最终只有最后一个订阅者能收到事件，完全违背了多播的初衷。

正确的实现思路：让Subject同时扮演Observable和Observer #

Subject的核心是自己维护订阅者集合，同时实现Observer的next/error/complete接口，当这些方法被调用时，遍历集合把事件转发给所有订阅者。另外，它作为Observable的订阅逻辑，就是把新的订阅者加入集合，并返回取消订阅的方法。

先完善你提到的Observer包装类（带状态管理和异常捕获）：

class Observer {
  constructor(next = () => {}, complete = () => {}, error = () => {}) {
    this._next = next;
    this._complete = complete;
    this._error = error;
    this.isCompleted = false;
  }

  _safeInvoke(fn, ...args) {
    if (!this.isCompleted) {
      try {
        fn(...args);
      } catch (err) {
        this.error(err);
        this.isCompleted = true;
      }
    }
  }

  next(...args) {
    this._safeInvoke(this._next, ...args);
  }

  complete() {
    this._safeInvoke(this._complete);
    this.isCompleted = true;
  }

  error(err) {
    this._safeInvoke(this._error, err);
    this.isCompleted = true;
  }
}

然后是符合RxJS设计思路的Subject实现：

class Subject extends Observable {
  constructor() {
    // Subject作为Observable的订阅逻辑：把订阅者加入集合
    super(observer => {
      this.subscribers.add(observer);
      // 返回取消订阅的方法：从集合中移除该订阅者
      return () => {
        if (!observer.isCompleted) {
          this.subscribers.delete(observer);
        }
      };
    });
    this.subscribers = new Set();
  }

  // 实现Observer接口：接收事件并广播给所有订阅者
  next(...args) {
    this.subscribers.forEach(subscriber => subscriber.next(...args));
  }

  error(err) {
    this.subscribers.forEach(subscriber => subscriber.error(err));
    // 发生错误后清理所有订阅者
    this.subscribers.clear();
  }

  complete() {
    this.subscribers.forEach(subscriber => subscriber.complete());
    this.subscribers.clear();
  }

  // 模仿EventEmitter的emit方法，本质就是调用next
  emit(...args) {
    this.next(...args);
  }
}

这个实现的核心逻辑解释 #

1. 继承Observable：Subject本身就是一个Observable，所以不需要额外创建内部Observable。订阅逻辑直接在super的构造函数里定义——每次有新订阅时，就把包装后的Observer加入subscribers集合，同时返回取消订阅的函数（从集合中删除）。
2. 实现Observer接口：next/error/complete方法负责把接收到的事件广播给所有订阅者，同时在error和complete后清理订阅者集合，避免内存泄漏。
3. emit方法：和Angular的EventEmitter一致，只是next方法的别名，方便习惯EventEmitter用法的开发者调用。

对比RxJS的实现逻辑 #

RxJS里的Subject核心思路和上面完全一致，只是做了更多细节优化：

• 用数组而非Set存储订阅者（方便批量操作）
• 加入了订阅状态管理（比如是否已完成/出错）
• 支持取消订阅的自动清理
• 衍生出了BehaviorSubject、ReplaySubject等变种，但基础的多播逻辑都是基于这个核心实现的。

这样实现的Subject就完全符合多播的需求，再也不用生硬地捕获Observer了~

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