javascript - ReactiveX:异步行为是如何实现的

Question

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

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

Answer 1

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

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

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

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

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

<script src="https://unpkg.com/rxjs@5/bundles/Rx.min.js"></script>

但是，如果指定了调度程序，将使用调度程序发出值(通常是异步的):

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; }

<script src="https://unpkg.com/rxjs@5/bundles/Rx.min.js"></script>

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

• AsyncScheduler 使用 setInterval ;
• AsapScheduler 使用 setImmediate ;和
• AnimationFrameScheduler 使用 requestAnimationFrame .

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