RXJS Subject异步订阅顺序问题：如何按订阅顺序同步执行回调？

问题根源 #

你的问题在于RxJS的Subject默认会同步遍历所有订阅者回调：当subject.next()触发时，会按订阅顺序依次调用每个回调，但类A的回调是async函数，执行到await时会立即返回一个Promise并让出主线程，此时主线程会继续执行类B的同步回调，导致B的日志先输出。

推荐方案：服务层控制串行执行 #

不需要修改类A和类B的代码，只需改造MyService，让所有订阅者的回调按订阅顺序串行执行，等待上一个回调完成（无论同步还是异步）后再执行下一个。这种方式完全保持A、B的独立性，符合你的需求。

改造后的MyService代码 #

import { Observable, Subscriber } from 'rxjs';

class MyService {
  // 维护订阅者的回调列表，按订阅顺序存储
  private readonly subscribers: Array<(value: any) => Promise<void> | void> = [];
  // 记录上一次执行的Promise，确保串行执行
  private lastExecution = Promise.resolve();

  public observe(): Observable<any> {
    return new Observable((subscriber) => {
      // 包装订阅者的next回调，处理异步返回值
      const callback = (value: any) => {
        try {
          const result = subscriber.next(value);
          // 如果回调返回Promise，直接返回；否则返回已resolved的Promise
          return result instanceof Promise ? result : Promise.resolve();
        } catch (err) {
          subscriber.error(err);
          return Promise.reject(err);
        }
      };

      // 添加到订阅者列表
      this.subscribers.push(callback);

      // 取消订阅时从列表移除
      return () => {
        const index = this.subscribers.indexOf(callback);
        if (index !== -1) {
          this.subscribers.splice(index, 1);
        }
      };
    });
  }

  // 替代原Subject的next方法，负责按顺序串行执行回调
  public next(value: any) {
    this.lastExecution = this.lastExecution.then(async () => {
      // 遍历所有订阅者，依次执行回调并等待完成
      for (const callback of this.subscribers) {
        await callback(value);
      }
    });
  }
}

工作原理 #

1. 自定义Observable来跟踪所有订阅者的回调，替代原生Subject的订阅逻辑。
2. 调用next()时，通过lastExecutionPromise链确保所有回调按订阅顺序串行执行：
   • 等待上一次所有回调执行完成后，再遍历当前订阅者列表。
   • 每个回调执行后，若返回Promise则等待其完成，再执行下一个回调。

不推荐方案：订阅者端同步阻塞异步操作 #

如果你一定要模拟C#中GetAwaiter().Wait()的同步阻塞效果，JavaScript中可以借助第三方库（如deasync）实现，但会阻塞主线程，仅适合Node.js测试场景，不推荐生产环境或浏览器使用：

类A的修改示例（Node.js环境） #

import deasync from 'deasync';

class A {
  constructor(private readonly service: MyService) {
    service.observe().subscribe( (value) => { 
      // 同步等待异步操作完成
      const promise = this.awaitAnOperation(value);
      deasync.loopWhile(() => !promise.isResolved());
      
      console.log('Class A subscription complete', value);
    });
  }

  private awaitAnOperation(value: any): Promise<void> {
    // 你的异步操作逻辑
    return new Promise(resolve => setTimeout(resolve, 1000));
  }
}

这种方式会卡住主线程直到Promise完成，严重影响性能和用户体验，因此优先选择服务层控制的方案。

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