在RxJS中如何正确使用asyncScheduler确认数据已完成处理？

你遇到的现象根源是asyncScheduler基于浏览器/Node环境的宏任务（底层用setTimeout实现），单元测试默认同步执行所有用例代码，用例执行结束时asyncScheduler调度的任务还在事件队列中未执行，所以看不到日志也验证不到逻辑。当你使用scheduled(events, asyncScheduler)时，事件本身的发射就被调度到宏任务队列，比直接用from同步发射的情况多了一层调度，更不会同步触发后续管道逻辑。

方案1：使用RxJS官方TestScheduler（最推荐） #

首先改造业务代码，把硬编码的asyncScheduler改为可注入的依赖，测试时替换为虚拟化时间的TestScheduler，可以精准控制任务执行时机，不需要真的等待异步时间：

// 业务代码改造示例
class YourService {
  // 支持外部传入调度器，默认用asyncScheduler
  constructor(private scheduler = asyncScheduler) {}

  yourBusinessMethod() {
    this.events$
      .pipe(
        observeOn(this.scheduler),
        // 其他原有操作符
      ).subscribe(anotherObservable$);
  }
}

单元测试代码示例：

import { TestScheduler } from 'rxjs/testing';

describe('YourService测试', () => {
  let testScheduler: TestScheduler;
  let service: YourService;

  beforeEach(() => {
    testScheduler = new TestScheduler((actual, expected) => {
      // 替换为你使用的测试框架的断言方法，比如Jest的toEqual
      expect(actual).toEqual(expected);
    });
    // 注入测试调度器
    service = new YourService(testScheduler);
  });

  it('应触发asyncScheduler对应的管道逻辑', () => {
    const testEvents = [/* 你构造的测试事件数组 */];
    // 同样用测试调度器构造输入流
    scheduled(testEvents, testScheduler).subscribe(service.events$);
    
    // 手动触发所有已调度的任务，所有管道逻辑会立即执行
    testScheduler.flush();

    // 这里添加你的断言，验证相关逻辑已被调用
  });
});

方案2：用fakeAsync模拟异步（适配Angular等框架场景） #

如果你的测试框架支持fakeAsync能力（比如Angular测试工具集），可以直接通过tick/flush触发所有宏任务：

import { fakeAsync, flush } from '@angular/core/testing';

it('应触发asyncScheduler对应的管道逻辑', fakeAsync(() => {
  const testEvents = [/* 你构造的测试事件数组 */];
  scheduled(testEvents, asyncScheduler).subscribe(service.events$);
  
  // 清空所有异步任务队列
  flush();

  // 执行断言
}));

方案3：用done回调等待异步执行（临时兼容方案） #

如果不想改现有代码也没有fakeAsync能力，可以用测试框架的异步回调机制等待宏任务执行：

it('应触发asyncScheduler对应的管道逻辑', (done) => {
  const testEvents = [/* 你构造的测试事件数组 */];
  scheduled(testEvents, asyncScheduler).subscribe(service.events$);

  // 放到下一个宏任务执行，等asyncScheduler的任务完成
  setTimeout(() => {
    // 执行断言
    done();
  }, 0);
});

该方案缺点是会拉长测试执行时间，不适合大量测试用例使用。

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