RxJS 如何创建或模拟异步？

Question

我在理解 RxJS 可观察对象和主题的执行模型/顺序时遇到问题。我读了很多关于 RxJS observables 的文献和博客文章，这是更好的承诺，因为它们的订阅可以取消，并且它们可以通过next()发出多个结果/值。

这个问题可能很容易回答，但是 RxJS 如何创建或模拟异步呢？RxJS Observables 是否包装 Promise 并创建一系列 Promise 以使代码执行异步？或者是因为实现的可观察模式，更改被异步传播到订阅者，但代码执行仍然是同步的？

在我看来，当 JavaScript 代码通过事件循环处理的任何 JavaScript 回调队列中的回调进行处理时，它是异步的。

Answer 1

在 RxJS 中，一切都与生产者有关。生产者可以是任何东西，它可以是同步的或异步的，因此 Observables 可以同步或异步发射。

让我们尝试了解什么是（a）同步行为。我将离开几个环节为主体的更深层次的理解：一个谈话由菲利普·罗伯茨，另一个谈话由Jake阿奇博尔德和Jake的博客，如果你不喜欢看长视频。

Tl;dw（atch）：所有 JavaScript 代码都是同步的，并在单个线程中执行。另一方面，可以从 JS 代码访问的 WebAPI 可能会在其他线程中执行一些其他内容并将结果带回 JavaScript 运行时。结果通过事件循环和回调传递到运行时。所以，当你说：

在我看来，javascript 代码在通过事件循环处理的任何 JavaScript 回调队列中的回调处理时是异步的。

你是对的。事件循环处理的回调是异步回调。具有异步回调的 WebAPI 示例有：setTimeoutand setInterval、DOM 事件、XHR 事件、Fetch 事件、Web 工作者、Web 套接字、Promise、MutationObserver 回调等。最后两个（Promises 和 MutationObservers）在不同的队列（微任务队列）上调度任务，但它仍然是异步的。

回到 RxJS。我已经说过，在 RxJS 中，它是关于生产者的一切。他们使用观察者包装生产者。引用文章中的 Ben Lesh：

[生产者] 是您用来获取值并将它们传递给observer.next(value).

这意味着同步的代码（和所有 JS 代码都是）在用 Observable 包装时将同步发出值。例如：

import { Observable } from 'rxjs';

const o = new Observable(observer => {
  [1, 2, 3].forEach(i => observer.next(i));
  observer.complete();
});

o.subscribe(x => console.log(x));

console.log('Anything logged after this?');

日志：

1
2
3
Anything logged after this?

另一方面，下一个示例使用setTimeout（它不是 ECMAScript 规范的一部分并使用异步回调）：

1
2
3
Anything logged after this?

记录这个：

Anything logged after this?
1

这意味着，即使我在 last 之前订阅了源 observable console.log，我们还是在observer发送下一个值之前收到了消息。这是因为setTimeout.

事实上，RxJS 有很多创建 Observable 的方法，这样你就不必通过包装所有这些来编写自己的实现。

因此，改进了第一个示例：

import { from } from 'rxjs'; 

from([1, 2, 3]).subscribe(i => console.log(i));

console.log('Anything logged after this?');

或改进的第二个例子：

import { of, scheduled, asyncScheduler } from 'rxjs'; 

scheduled(of(1), asyncScheduler).subscribe(i => console.log(i));

console.log('Anything logged after this?');

scheduled创建运算符使用调度程序在不同的任务队列上调度事件。asyncScheduler内部用于setTimeout将事件分派到宏任务队列，而asapScheduler内部使用 Promises，因为它使用微任务队列。

然而，setTimeout是异步行为最明显和最重复的例子。XHR 对我们来说更有趣。Angular 的HTTP客户端执行与我在前两个示例中所做的相同的包装，因此，当响应到来时，它被传输到responseObserverusing next。

当响应来自服务器时，XMLHttpRequest对象将其放入宏任务队列，一旦调用堆栈被清除，该队列就会被事件循环推送到调用堆栈，并且可以将消息传递给responseObserver.

这样，异步事件发生，并且包装该XMLHttpRequest对象的 Observable 的订阅者异步获取它们的值。

我阅读了很多关于 RxJS observables 是更好的承诺的文献和博客文章，因为它们的订阅可以被取消并且它们可以通过 next() 发出多个结果/值。

Observables 和 Promises 的区别确实在于 Observables 是可取消的。当您使用 WebAPI 进行大量工作时，这是最重要的，因为它们中的许多都需要具有可取消的方法（以便在我们停止使用它们时不会丢失资源）。

事实上，由于 RxJS 有很多创建操作符封装了很多 WebAPI，它们已经在为你处理取消操作了。您所要做的就是跟踪订阅并在适当的时候取消订阅。可以在此处找到可能对此有所帮助的文章。

RxJS Observables 是否环绕 promise 并创建一系列 promise 以使代码执行异步？

不，他们包装了一个生产者。任何可以调用observer.next方法的东西。如果生产者使用调用observer.next方法的异步回调，则 Observables 异步发射。反过来，它们同步发射。

但是，即使原始发射是同步的，也可以通过使用调度程序将它们分派为异步发射。