ReactiveX：异步行为是如何实现的答案

【问题标题】：ReactiveX: How is Asynchronous Behavior AchievedReactiveX：异步行为是如何实现的
【发布时间】：2018-01-19 08:26:56
【问题描述】：

RxJS Observables 是可以随时间同步或异步返回零到无限值的函数。

但是异步行为究竟是如何实现的呢？它以某种方式使用单独的线程，还是使用 Web API 和 JS 事件循环？

【问题讨论】：

我猜它使用 JS 事件 - 类似于 window.DispatchEvent -> addEventListener。
我猜 Ajax 调用是异步的，或者是数据库调用，或者是文件读取（在 Nodejs 中）
我相信 AJAX 通过使用各种内置的浏览器 API（又名 Web API）来实现异步行为，例如 XMLHttpRequest。这些 API 内置在浏览器中，因此它们在与主执行线程不同的线程中处理。当这些函数准备好执行时，Web API 将它们推送到事件循环，一旦调用堆栈为空，事件循环又将函数推送到调用堆栈。我不确定 RxJS 是否使用像这样的内置浏览器 API。

【解决方案1】：

异步行为源于运行时环境的 API。

例如，implementation of fromEvent 调用addEventListener，implementation of AjaxObservable 创建一个XMLHttpRequest。

RxJS 可以使用调度程序将异步行为引入其他同步的可观察对象。例如，使用 of 创建的 observable 同步发出：

Rx.Observable
  .of(1, 2, 3)
  .subscribe(value => console.log("emitted", value));
console.log("subscribed");

.as-console-wrapper { max-height: 100% !important; top: 0; }

&lt;script src="https://unpkg.com/rxjs@5/bundles/Rx.min.js"&gt;&lt;/script&gt;

但是，如果指定了调度程序，则将使用调度程序发出值（通常是异步的）：

Rx.Observable
  .of(1, 2, 3, Rx.Scheduler.asap)
  .subscribe(value => console.log("emitted", value));
console.log("subscribed");

.as-console-wrapper { max-height: 100% !important; top: 0; }

&lt;script src="https://unpkg.com/rxjs@5/bundles/Rx.min.js"&gt;&lt;/script&gt;

RxJS 中有几个调度器，它们都是使用异步浏览器 API 实现的：

当然，RxJS 也可以与 Node.js 一起使用，并且在那种环境中，使用 Node 的 API。

【讨论】：