RxJS mergeAll

将高阶 Observable 转换为其中的一阶 Observable 同时传递内部 Observable 上发出的所有值。

mergeAll<T>(concurrent: number = Number.POSITIVE_INFINITY): OperatorFunction<ObservableInput<T>, T>

参量

同时	可选的。 默认值为 Number.POSITIVE_INFINITY。 内部最大数量 可观察项被同时订阅。

returns

OperatorFunction<ObservableInput<T>, T>：发出所有值的 Observable 源 Observable 发出的内部Observable。

描述

展平可观察的事物。

mergeAll订阅发出 Observable 的 Observable，也称为 一个更高阶的 Observable。 每次观察到其中一种发射出的内部 可观察对象，它订阅并从 内部 Observable 在输出 Observable上。 仅输出可观察 一旦所有内部可观测对象都完成，则完成。 传递的任何错误 内部 Observable 将立即在输出 Observable 上发出。

例子

为每个点击事件产生一个新的 Observable 间隔，并将其输出混合为一个 Observable

import { fromEvent, interval } from 'rxjs';
import { map, mergeAll } from 'rxjs/operators';


const clicks = fromEvent(document, 'click');
const higherOrder = clicks.pipe(map((ev) => interval(1000)));
const firstOrder = higherOrder.pipe(mergeAll());
firstOrder.subscribe(x => console.log(x));

每次点击从每秒 0 到 9 进行计数，但仅允许 2 个并发计时器

import { fromEvent, interval } from 'rxjs';
import { take, map, mergeAll } from 'rxjs/operators';


const clicks = fromEvent(document, 'click');
const higherOrder = clicks.pipe(
  map((ev) => interval(1000).pipe(take(10))),
);
const firstOrder = higherOrder.pipe(mergeAll(2));
firstOrder.subscribe(x => console.log(x));

也可以看看

• combineAll
• concatAll
• exhaust
• merge
• mergeMap
• mergeMapTo
• mergeScan
• switchAll
• switchMap
• zipAll