typescript - Rxjs Subject#next 是否异步？

Question

您好，我想知道以下代码是否按顺序执行？即日志记录“Worker has finished task C”是否总是发生在日志记录“Finished 3 task(s)”之后？

长问题:使用scan 运算符，我可以确定任务是按顺序执行的，我并不担心。让我担心的是，我希望最后一个订阅仅在任务 C 完成后才做某事，我不确定放置 o.complete() 的位置是否能保证这一点。例如将 start() 运行 do.("A") -> do.("B") -> do.("C") 而无需等待扫描完成并立即运行 o.complete()，给出输出:

Worker has finished task C
Doing task A
Finished 1 task(s)
Doing task B
Finished 2 task(s)
Doing task C
Finished 3 task(s)

如果是这种情况，您如何修复代码以实现我所描述的？

https://stackblitz.com/edit/typescript-xhhwme

class Worker {
  private tasks: Subject<string>;
  public init(): Observable<number> {
    this.tasks = new Subject<string>();
    return this.tasks.scan((count, task) => {
      console.log("Doing task " + task);
      return ++count;
    }, 0).asObservable();
  }
  public do(task: string): void {
    this.tasks.next(task);
  }
}

function start(worker: Worker): Observable<void> {
  return Observable.create(o => {
    const monitor = worker.init();
    monitor.subscribe(c => console.log("Finished " + c + " task(s)"));
    worker.do("A");
    worker.do("B");
    worker.do("C");
    o.complete();
    worker.do("D");
  });
}

const worker = new Worker();
start(worker).subscribe({
  complete: () => console.log("Worker has finished task C")
});

Answer 1

TLDR:Subject.next 是同步的。

如果源是同步的，则 react 流是同步的，除非您明确将它们设为异步或将它们与异步流混合。这些都不会发生在您的代码中。一些例子:

//Synchronous
of(1,2)
  .subscribe(console.log);

//asynchronous because of async source
interval(1000)
  .subscribe(console.log);

//aynchronous because one stream is async (interval)
of(1,2)
  .pipe(
    mergeMap(x => interval(1000).pipe(take(2)))
  )
  .subscribe(console.log);

//async because we make it async
of(1,2, asyncScheduler)
  .subscribe(console.log);

您的示例中发生了什么？ Observable.create 中的所有内容都将立即执行。当您调用 worker.do("A"); 时，this.tasks.next(task); 发出一个新值，tasks流链被执行(同步)。 B 和 C 也是如此。

当您调用 o.complete(); 时，start(worker) 流完成并打印 "Worker has finished task C" .然后 D 由 tasks 流执行。

您可以在这些文章中找到有关异步/同步行为的更多详细信息:

• Concurrency and Asynchronous Behavior with RxJS
• Reactive Streams and Multithreading