RxJS如何创建或模拟异步? [英] How does RxJS create or simulate asynchronism?

问题描述

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

I have problems understand the execution model/order of RxJS Observables and Subjects. I read a lot of literature and blog posts about RxJS observables being the better promise since their subscription can be canceled and they can emit multiple results/values via next().

这个问题可能很容易回答，但是RxJS如何创建或模拟异步?RxJS Observables是否会包装Promise并创建Promise序列以使代码执行异步?还是因为实现了可观察的模式，更改才能异步传播给订户，但是代码执行仍然是同步的?

This question might be answered easily but how does RxJS create or simulate asynchronism? Does RxJS Observables wrap around promises and create a sequence of promises to make the code execution asynchronous? Or is it because of the implemented observable pattern that change is propagated asynchronous to subscribers but code execution is still synchronous?

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

In my point of view javascript code is asynchronous when it is handle via callbacks in any of the JavaScript callback queues processed by the event loop.

推荐答案

在RxJS中，一切都与生产者有关.生产者可以是任何事物，并且可以是同步的或异步的，因此Observable可以同步或异步地发出.

In RxJS, everything is about producer. The producer can be anything and it can be synchronous or asynchronous, thus Observables can both emit synchronously or asynchronously.

让我们尝试理解什么是(a)同步行为.我将留下几个链接以使您对该主题有更深入的了解:Philip Roberts的对话，由Jake Archibald和Jake的

Lets try to understand what is (a)synchronous behavior. I will leave couple of links for deeper understanding of the subject: a talk by Philip Roberts, another talk by Jake Archibald and Jake's blog if you don't like watching long videos.

Tl; dw (补丁):所有JavaScript代码都是同步的，并且在单个线程中执行.另一方面，可以从JS代码访问的WebAPI可以在其他线程中执行其他一些操作，并将结果返回给JavaScript运行时.结果通过事件循环和回调传递到运行时.所以，当你说:

Tl;dw(atch): all JavaScript code is synchronous and executes within a single thread. On the other hand, WebAPIs, which can be accessed from JS code, may execute some other stuff in other threads and bring back the result to the JavaScript runtime. And the results are passed through to the runtime by Event loop and the callbacks. So, when you say:

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

In my point of view javascript code is asynchronous when it is handle via callbacks in any of the JavaScript callback queues processed by the event loop.

您是对的.事件循环处理的回调是异步回调.具有异步回调的WebAPI的示例包括: setTimeout 和 setInterval ，DOM事件，XHR事件，Fetch事件，Web工作者，Web套接字，Promises，MutationObserver回调等等.最后两个(Promises和MutationObservers)将任务安排在不同的队列(微任务队列)上，但仍然是异步的.

You're right. A callback handled by the Event loop is the asynchronous callback. Examples of WebAPIs which have asynchronous callbacks are: setTimeout and setInterval, DOM events, XHR events, Fetch events, Web workers, Web sockets, Promises, MutationObserver callbacks and so on. The last two (Promises and MutationObservers) schedule tasks on a different queue (microtask queue), but it's still asynchronous.

返回RxJS.我已经说过，在RxJS中，这一切都是关于生产者的.他们使用观察者包装生产者.引用文章中的Ben Lesh:

Back to RxJS. I already told that in RxJS it's everything about the producer. They wrap producers using observers. To quote Ben Lesh from the article:

[生产者]是您用来获取值并将其传递给 observer.next(value)的任何东西.

这意味着同步的代码(以及所有JS代码都是)在被Observable包裹时将同步发出值.例如:

This means that the code that is synchronous (and all JS code is) will synchronously emit values when wrapped with an Observable. For example:

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规范的一部分，并且使用异步回调):

On the other hand, next example uses setTimeout (which is not part of the ECMAScript specification and uses asynchronous callback):

import { Observable } from 'rxjs';

const o = new Observable(observer => {
  setTimeout(() => {
    observer.next(1);
    observer.complete();
  }, 0);
});

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

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

记录以下内容:

Anything logged after this?
1

这意味着，即使我在上一个 console.log 之前订阅了可观察的源，在 observer 发送下一个值之前，我们仍然收到了消息.这是因为 setTimeout 的异步特性.

This means that, even though I subscribed to the source observable before last console.log, we've got the message before observer sent next value. This is because of the asynchronous nature of setTimeout.

实际上，RxJS有许多创建Observable的方法，因此您不必通过包装所有这些来编写自己的实现.

In fact, RxJS has many ways of creating Observables so that you don't have to write your own implementations by wrapping all of this.

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

So, improved first example:

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?');

计划的创建操作员使用计划的人用于在不同任务队列上调度事件. asyncScheduler 内部使用 setTimeout 将事件调度到宏任务队列，而 asapScheduler 内部使用Promises，因为它使用微任务队列.

scheduled creation operator uses schedulers for dispatching events on different task queues. asyncScheduler internally uses setTimeout to dispatch the event to the macrotask queue, while asapScheduler internally uses Promises as it uses microtask queue.

但是， setTimeout 是异步行为的最明显和最重复的示例.XHR对我们来说更有趣.Angular的 HTTP 客户端不会与前两个示例中的包装相同，因此，当响应到来时，将使用 next 将其传输到 responseObserver .

However, setTimeout is the most obvious and the most repeated example of asynchronous behavior. XHR is the one that is much more interesting to us. Angular's HTTP client does the same wrapping as I did in my first two examples, so that, when response comes, it is transferred to the responseObserver using next.

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

When the response comes from the server, XMLHttpRequest object puts it to macrotask queue which gets pushed to the call stack by Event loop once call stack is cleared, and the message can be passed to the responseObserver.

这样，异步事件就会发生，并且包装了 XMLHttpRequest 对象的Observable的订阅者将异步获取其值.

This way, the asynchronous event happens, and the subscribers to the Observable that wraps that XMLHttpRequest object get their value asynchronously.

我阅读了很多有关RxJS可观察对象的更好的文献和博客文章，因为它们的订阅可以取消，并且可以通过next()发出多个结果/值.

I read a lot of literature and blog posts about RxJS observables being the better promise since their subscription can be canceled and they can emit multiple results/values via next().

Observables和Promise之间的区别确实在于，Observables是可取消的.当您经常使用WebAPI时，这是最重要的，因为它们中的许多都需要有可取消的方法(这样，当我们停止使用它们时，资源不会丢失).

The difference between Observables and Promises is indeed in the fact that Observables are cancelable. This is the most important when you're working a lot with WebAPIs as many of them need to have means to be cancelable (so that resources are not lost when we stop using them).

实际上，由于RxJS具有许多包装了许多WebAPI的创建运算符，因此它们已经在为您处理取消事务.您要做的就是跟踪订阅并在适当的时候取消订阅.可能对此有所帮助的文章可以在此处找到.

In fact, since RxJS has many creation operators that wrap many of the WebAPIs, they're already dealing with the cancelation stuff for you. All you have to do is to keep track of the subscriptions and to unsubscribe at the right moment. Article that might be helpful for that can be found here.

RxJS Observables是否会包装Promise并创建一个Promise序列以使代码执行异步?

Does RxJS Observables wrap around promises and create a sequence of promises to make the code execution asynchronous?

不，他们包装了生产者.任何可以调用 observer.next 方法的东西.如果生产者使用调用 observer.next 方法的异步回调，则Observable异步发出.相反，它们会同步发射.

No, they wrap a producer. Anything that can call observer.next method. If a producer uses asynchronous callbacks which call observer.next method, then Observables emit asynchronously. Other way around, they emit synchronously.

但是，即使原始排放是同步的，也可以使用调度程序将它们调度为异步排放.

But, even though original emissions are synchronous, they can be dispatched to be emitted asynchronously by using schedulers.

这篇关于RxJS如何创建或模拟异步?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！