RXJS:从promises创建的串联可观测对象的非顺序行为

基础概念

RxJS（Reactive Extensions for JavaScript）是一个用于处理异步数据流的库。它使用可观察对象（Observables）来表示数据流，并提供了多种操作符来处理这些数据流。Promises 是 JavaScript 中用于处理异步操作的一种对象，它表示一个异步操作的最终完成（或失败）及其结果值。

类型

从 Promises 创建的串联可观测对象属于 RxJS 中的 from 操作符。这个操作符可以将一个 Promise 转换为一个 Observable。

应用场景

当你有一个异步操作（例如网络请求）并且希望使用 RxJS 的操作符来处理这个异步操作的结果时，可以使用 from 操作符。

非顺序行为的原因及解决方法

非顺序行为的原因

从 Promises 创建的串联可观测对象的非顺序行为通常是由于多个异步操作并发执行导致的。例如，如果你有多个网络请求，并且这些请求是并发执行的，那么它们的结果可能会以非顺序的方式到达。

解决方法

使用 concatMap 操作符：concatMap 操作符会确保每个源 Observable 发出的值都按顺序传递给嵌套的 Observable。这样可以保证结果的顺序性。

import { from } from 'rxjs';
import { concatMap } from 'rxjs/operators';

const promises = [
  Promise.resolve(1),
  Promise.resolve(2),
  Promise.resolve(3)
];

from(promises)
  .pipe(
    concatMap(promise => from(promise))
  )
  .subscribe(value => console.log(value));

使用 mergeMap 操作符并设置并发数：如果你希望并发执行多个异步操作，但仍然希望控制并发数，可以使用 mergeMap 操作符并设置 concurrency 参数。

import { from } from 'rxjs';
import { mergeMap } from 'rxjs/operators';

const promises = [
  Promise.resolve(1),
  Promise.resolve(2),
  Promise.resolve(3)
];

from(promises)
  .pipe(
    mergeMap(promise => from(promise), 1) // 设置并发数为 1
  )
  .subscribe(value => console.log(value));