RxJS switchMap未取消内部合并Observable问题咨询（v5.5.6）

嘿，我来帮你搞清楚这个问题～首先咱们先把场景和现象理清楚：

问题场景与异常现象 #

你提到在RxJS 5.5.6中，使用switchMap时，当第二个源Observable发出值后，虽然switchMap已经取消了对第一个内部Observable的订阅，但第一个内部Observable居然还在运行——比如日志里持续输出do: a0 .. b0 .. c0，这和预期不符：咱们预期switchMap切换时，前一个内部Observable应该完全终止才对。

举个典型的有问题的代码示例（模拟你的场景）：

import { interval, from } from 'rxjs';
import { switchMap, tap } from 'rxjs/operators';

// 有问题的内部Observable：未处理取消订阅后的清理
const createInnerStream = (id: string) => interval(1000).pipe(
  tap(num => console.log(`do: ${id}${num}`))
  // 这里如果是自定义Observable，可能忘了返回清理逻辑
);

// 源Observable：先发出'a'，过一会儿发出'b'
from(['a', 'b']).pipe(
  switchMap(id => createInnerStream(id))
).subscribe();

运行后你会发现，当切换到b对应的Observable后，a的interval日志还在持续输出，这就是你遇到的异常。

预期输出的修正代码示例 #

你提到用另一段代码得到了预期结果，大概率是给内部Observable加上了正确的终止逻辑，比如这样：

import { interval, from } from 'rxjs';
import { switchMap, tap, takeUntil, finalize } from 'rxjs/operators';
import { Subject } from 'rxjs/Subject';

const createInnerStream = (id: string) => {
  const stopSignal$ = new Subject();
  return interval(1000).pipe(
    tap(num => console.log(`do: ${id}${num}`)),
    takeUntil(stopSignal$),
    // 当Observable终止时触发清理
    finalize(() => {
      stopSignal$.next();
      stopSignal$.complete();
      console.log(`内部Observable ${id} 已终止`);
    })
  );
};

from(['a', 'b']).pipe(
  switchMap(id => createInnerStream(id))
).subscribe();

或者如果是自定义Observable，要确保返回清理函数：

const createInnerStream = (id: string) => new Observable(observer => {
  let num = 0;
  const timer = setInterval(() => {
    observer.next(num++);
    console.log(`do: ${id}${num}`);
  }, 1000);
  
  // 关键：返回清理函数，取消订阅时会执行
  return () => {
    clearInterval(timer);
    console.log(`内部Observable ${id} 已终止`);
  };
});

问题根源与解释 #

其实核心问题不是switchMap没干活——switchMap确实会在切换时调用前一个内部Observable的unsubscribe()方法。但RxJS的Observable遵循一个规范：只有当Observable自身提供了清理逻辑时，取消订阅才能真正终止内部的异步操作。

换句话说：

1. 如果你的内部Observable是类似interval这种官方操作符生成的，其实它本身是带有清理逻辑的（取消订阅会清除定时器），但如果你的场景里是自定义Observable，忘了返回清理函数（比如上面例子里的clearInterval），那定时器就会一直跑，日志自然停不下来。
2. 有些复杂的流组合里，可能内部Observable依赖了其他未被终止的流，这时候需要用takeUntil、finalize这类操作符来手动绑定终止信号，确保取消订阅时能斩断所有异步任务。

你的修正代码之所以生效，就是因为它给内部Observable补上了必要的清理逻辑，让switchMap的取消订阅操作能真正触发内部任务的终止，而不是只停留在“取消订阅”这个动作本身。

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