RxJS Observable结合async函数与toPromise的异常行为问询

咱们一步步拆解你遇到的这个奇怪现象，核心问题出在RxJS的冷Observable特性和toPromise()的行为逻辑上：

1. 冷Observable的「每次订阅都重启数据源」特性 #

RxJS里默认的Observable是冷Observable——简单说就是，每一次订阅（包括调用toPromise()，因为它内部会偷偷发起订阅），都会重新执行一遍Observable创建函数里的代码。

看你定义的Observable：

const x = new Observable((o) => {
  setInterval(() => o.next(++i), 1000);
});

当你执行x.subscribe(...)时，会运行这个回调，创建一个定时器，每秒递增全局变量i并发送值；而当你调用x.toPromise()时，又会再跑一遍这个回调，创建第二个完全独立的定时器，同样会递增同一个全局i。

这就导致i被两个定时器同时修改：每过1秒，两个定时器各执行一次++i，所以i每次会跳着加2，而不是1。

2. toPromise()的「无限等待」坑 #

在你的async/await循环里：

(async () => {
  while (true) {
    try {
      console.log("loop");
      console.log("value", await x.toPromise());
    } catch (e) {
      console.log(e);
    }
  }
})();

旧版RxJS的toPromise()逻辑是：必须等Observable发出complete通知，才会把最后一个发出的值resolve给你。但你的Observable是无限的（从来没调用o.complete()），所以await x.toPromise()会一直卡着，永远不会resolve。这就导致while(true)循环只跑了第一次，loop只打印一次，后面的代码再也没机会执行。

你用递归then的写法也会遇到同样的问题：每次调用a()都会新增一个订阅（也就是新增一个定时器），但toPromise()永远不会resolve，所以then里的代码永远不会执行，递归也就不会触发，同时还会不断堆积定时器，让i的递增速度越来越快。

3. 为什么subscribe只打印偶数？ #

再理一遍代码的执行顺序：

1. 先执行async函数，进入循环后调用x.toPromise()，发起第一个订阅，创建定时器A；
2. await卡住async函数，然后执行后面的x.subscribe(...)，发起第二个订阅，创建定时器B；
3. 两个定时器每秒触发一次：
   • 定时器A先触发：++i把i变成1，发送值1（但这个值被toPromise的订阅接收，却因为没有complete，永远不会被打印）；
   • 紧接着定时器B触发：++i把i变成2，发送值2（这个值被你的subscribe回调接收，所以打印subscribe 2）；
   • 下一轮循环，定时器A把i变成3，定时器B把i变成4，subscribe打印subscribe 4；
   • 以此类推，就出现了你看到的偶数输出。

怎么修复？ #

如果想让所有订阅共享同一个数据源，用share()操作符把冷Observable转成热的：

const x = new Observable((o) => {
  setInterval(() => o.next(++i), 1000);
}).pipe(share());

这样不管订阅多少次，只会创建一个定时器，i每次递增1，subscribe会收到1、2、3...

如果想让toPromise()能拿到值，建议结合first()操作符（只取第一个值就完成）：

console.log("value", await x.pipe(first()).toPromise());

另外，如果你用的是RxJS 7+，推荐用firstValueFrom或lastValueFrom替代toPromise()，它们的行为更明确，避免踩坑：

import { firstValueFrom } from 'rxjs';
// ...
console.log("value", await firstValueFrom(x));

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