如何将Observable管道到Subject且避免不必要的热启动？

问题本质 #

原代码的问题在于提前手动订阅了mouseMove$：fromEvent返回的是冷Observable，仅在被订阅时才会绑定DOM事件监听、退订时移除监听。提前订阅后，mousemove的事件监听和后续superExpensiveComputation的执行完全和subject$的订阅状态解耦，哪怕没有订阅者监听subject$，昂贵计算也会持续执行。

解决方案 #

分两种场景对应不同实现：

场景1：必须复用已有的subject$实例 #

如果现有代码已经大量依赖这个subject$，不方便替换，可以劫持subject$的订阅/退订逻辑，把mouseMove$的生命周期和subject$的订阅者数量绑定：

import { Subject, fromEvent } from 'rxjs';
import { map } from 'rxjs/operators';

const subject$ = new Subject();
const mouseMove$ = fromEvent(document, 'mousemove').pipe(
  map(it => superExpensiveComputation(it))
);

// 保存源Observable的订阅引用，用于后续退订
let sourceSubscription = null;
const originalSubscribe = subject$.subscribe.bind(subject$);

// 重写subject的订阅逻辑
subject$.subscribe = (...args) => {
  // 第一个订阅者接入时，才订阅源Observable
  if (subject$.observers.length === 0 && !sourceSubscription) {
    sourceSubscription = mouseMove$.subscribe(val => subject$.next(val));
  }
  const subscription = originalSubscribe(...args);
  const originalUnsubscribe = subscription.unsubscribe.bind(subscription);
  
  // 重写当前订阅的退订逻辑
  subscription.unsubscribe = () => {
    originalUnsubscribe();
    // 所有订阅者都退订时，取消源的订阅，移除事件监听，停止昂贵计算
    if (subject$.observers.length === 0 && sourceSubscription) {
      sourceSubscription.unsubscribe();
      sourceSubscription = null;
    }
  };
  return subscription;
};

该方案下mouseMove$的订阅、退订完全和subject$的订阅者状态同步，不会出现无意义的昂贵计算。

场景2：无历史包袱，可直接替换原有实现 #

如果不需要保留外部的subject$实例，直接用RxJS内置的share操作符即可，它是多播+引用计数的封装，底层默认用Subject实现多播，完全满足需求：

import { fromEvent } from 'rxjs';
import { map, share } from 'rxjs/operators';

// 直接生成共享的多播Observable，替代原有subject$ + mouseMove$的组合
const sharedMouseMove$ = fromEvent(document, 'mousemove').pipe(
  map(it => superExpensiveComputation(it)),
  share()
);

该实现的效果：

• 第一个订阅者接入时才会绑定mousemove监听、触发计算逻辑
• 所有订阅者共享同一份事件监听和计算结果
• 最后一个订阅者退订时自动移除事件监听，不会再执行昂贵计算

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