异步回调创建Observable：代理与订阅取消转发的实现方案

1. 当前使用Subject的方式是否正确？ #

答案是不正确，核心问题有三个：

• 资源浪费：create函数一被调用就会执行mkAsync，不管有没有订阅者订阅返回的Subject。这意味着即使没人关心异步结果，setTimeout和后续的Observable创建、订阅都会执行，完全是不必要的开销。
• 多播不符合预期：Subject是多播对象，所有订阅它的观察者都会共享同一个底层的Observable.of('some result')流。但从你的代码逻辑来看，大概率是希望每个订阅都独立触发一次异步操作（否则没必要传入delay参数），当前实现做不到这一点。
• 取消订阅不转发：Subject的unsubscribe只会停止向自身的观察者推送数据，不会关联到底层Observable的Subscription，所以取消订阅Subject后，底层的异步逻辑还是会继续执行，这正是你遇到的核心问题。

2. 有没有不使用Subject的替代方案？ #

当然有，而且这才是RxJS推荐的标准做法：用new Observable()（RxJS 6+的写法）创建自定义Observable，完全不需要Subject。这种方式可以让你在订阅发生时才触发异步逻辑，而且天然支持清理操作，完美适配你的场景。

举个优化后的示例：

import { Observable, of } from 'rxjs';

const mkAsync = (observer, delay) => {
  let innerSubscription;
  // 启动定时器
  const timerId = setTimeout(() => {
    // 订阅底层Observable并保存引用
    innerSubscription = of('some result').subscribe(observer);
  }, delay);

  // 返回清理函数：取消定时器 + 取消底层Observable订阅
  return () => {
    clearTimeout(timerId);
    if (innerSubscription) innerSubscription.unsubscribe();
  };
};

const create = arg => {
  return new Observable(observer => {
    // 当有人订阅这个Observable时，才执行异步逻辑
    return mkAsync(observer, arg);
  });
};

这个方案里，只有当调用create()返回的Observable被订阅时，才会触发mkAsync；一旦取消订阅，清理函数会自动执行，同时取消定时器和底层Observable的订阅，完全符合你的需求。

3. 如何确保当Subject取消订阅时，底层Observable也能被取消？ #

如果因为某些限制必须保留Subject（非常不推荐，因为会引入不必要的复杂度），你需要手动实现引用计数来跟踪订阅者数量，当最后一个订阅者取消订阅时，主动取消底层的Subscription。

具体实现代码如下：

import { Subject, of } from 'rxjs';

const mkAsync = (observer, delay) => {
  let innerSubscription;
  const timerId = setTimeout(() => {
    innerSubscription = of('some result').subscribe(observer);
  }, delay);

  return () => {
    clearTimeout(timerId);
    if (innerSubscription) innerSubscription.unsubscribe();
  };
};

const create = arg => {
  const subject = new Subject();
  let subscriberCount = 0;
  let cleanupFn = null;

  // 重写Subject的subscribe方法，添加引用计数逻辑
  const originalSubscribe = subject.subscribe.bind(subject);
  subject.subscribe = (...args) => {
    const sub = originalSubscribe(...args);
    subscriberCount++;
    
    // 第一个订阅者触发异步逻辑
    if (subscriberCount === 1) {
      cleanupFn = mkAsync(subject, arg);
    }

    // 重写订阅的unsubscribe方法，更新计数并清理资源
    const originalUnsubscribe = sub.unsubscribe.bind(sub);
    sub.unsubscribe = () => {
      originalUnsubscribe();
      subscriberCount--;
      
      // 最后一个订阅者取消时，清理底层资源
      if (subscriberCount === 0 && cleanupFn) {
        cleanupFn();
        cleanupFn = null;
      }
    };
    return sub;
  };

  return subject;
};

不过要注意，这种方式属于hack性质的修改，不如直接用new Observable()来得优雅，而且需要小心处理计数逻辑，避免内存泄漏。

内容的提问来源于stack exchange，提问作者André Rüdiger