如何在Observable中使用async/await？无法触发unsubscribe清除定时器

嘿，我之前也踩过这个坑！问题的核心在于你把Observable构造函数的回调写成了async函数，这直接打乱了RxJS的清理逻辑机制。

为什么会出现这个问题？ #

Observable的构造函数期望它的回调函数直接返回一个清理函数（或者Subscription对象），当你调用unsubscribe()时，RxJS会执行这个返回的函数来清理资源（比如你的定时器）。但如果回调是async函数，它的返回值会自动被包装成一个Promise，而不是你写的那个return () => {...}清理函数。RxJS无法识别这个Promise作为清理逻辑，所以当你调用unsubscribe()时，根本不会执行你的定时器清除代码。

修复方案一：移除async，在回调内部处理异步逻辑 #

把async关键字去掉，回调直接返回清理函数，异步逻辑放在内部用then处理：

const { Observable } = require("rxjs");

const test = () => new Observable((subscriber) => {
  let intervalId;
  
  // 内部处理异步逻辑，不把整个回调变成async
  Promise.resolve().then(() => {
    intervalId = setInterval(() => {
      subscriber.next(Math.random());
      console.log("zz");
    }, 500);
  });

  // 返回真正的清理函数，确保能拿到定时器ID
  return () => {
    console.log("asdsad");
    if (intervalId) {
      clearInterval(intervalId);
    }
  };
});

const xyz = test().subscribe(console.log);
setTimeout(() => {
  xyz.unsubscribe();
}, 3000);

修复方案二：用IIFE包装async/await（如果一定要用） #

如果你的异步逻辑必须用async/await，可以在回调内部用立即执行函数表达式（IIFE）来包裹，但回调本身不能是async：

const { Observable } = require("rxjs");

const test = () => new Observable((subscriber) => {
  let intervalId;
  
  // 用IIFE处理async逻辑，回调本身保持同步
  (async () => {
    await Promise.resolve();
    intervalId = setInterval(() => {
      subscriber.next(Math.random());
      console.log("zz");
    }, 500);
  })();

  return () => {
    console.log("asdsad");
    if (intervalId) clearInterval(intervalId);
  };
});

const xyz = test().subscribe(console.log);
setTimeout(() => {
  xyz.unsubscribe();
}, 3000);

更推荐的方案：用RxJS操作符代替手动创建Observable #

其实手动写Observable构造函数是比较底层的做法，RxJS提供了丰富的操作符来帮你处理异步逻辑，同时自动管理资源清理。比如用from和switchMap组合：

const { from, interval } = require("rxjs");
const { map, switchMap } = require("rxjs/operators");

const test = () => from(Promise.resolve()).pipe(
  switchMap(() => interval(500).pipe(
    map(() => Math.random())
  ))
);

const xyz = test().subscribe(console.log);
setTimeout(() => {
  xyz.unsubscribe();
}, 3000);

这个方案里，interval的订阅会被RxJS自动管理，当你调用unsubscribe()时，RxJS会自动清除定时器，完全不需要手动写clearInterval。

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