合并固定数量的mergeMapped observables - RxJS

Question

我正在使用RxJS v6，但这个问题也适用于v5。

当使用mergeMap时，我的原始数组消失了，虽然我可以并行执行多个操作，但是当我发送到mergeMap的所有可观察数据都完成时，我不再有监视方法。

Example

of([1, 2, 3, 4])
.pipe(
    mergeMap(values => values),
)
.subscribe(console.log)

// 1
// 2
// 3
// 4

我想看：

// [1, 2, 3, 4]

到目前为止，我提出的唯一方法是获得数组的长度，但我确信必须有一些我缺少的运算符：

of([1, 2, 3, 4])
.pipe(
    switchMap(values => (
        of(values)
        .pipe(
            mergeMap(value => value),
            bufferCount(values.length),
        )
    ))
)
.subscribe(console.log)

Answer 1

使用mergeMap时，我的原始数组消失了

原因是mergeMap接受ObservableInput作为你传递的函数的参数.javascript Array是ObservableInput因此在mergeMap工作，mergeMap完成它的工作，这是为了使ObservableInput变平（考虑到mergeMap之前被称为flatMap）。

所以，正如@cartant所说，如果你想回到一个数组，你必须使用toArray运算符。换一种说法

of([1, 2, 3, 4])
.pipe(
    mergeMap(value => { // do stuff with a value of an array}),
    toArray()
)
.subscribe(console.log)

相当于

of([1, 2, 3, 4])
.pipe(
    map(values => values.map(value => { // do stuff with a value of an array})),
)

如果您的数组虽然包含Observables，并且您希望最终获得它们在所有它们发出时通知的值，那么您必须使用forkJoin。这是一个简单的例子

of([1, 2, 3, 4].map(n => of(n)))
.pipe(
    switchMap(observablesOfValues => forkJoin(observablesOfValues))
)
.subscribe(console.log)