我怎样才能一次只有 X 个 observables 处于活动状态？

Question

我使用的是 RxJS v6，但答案也适用于 RxJS v5。

我想这样做，如果我说 8 个值，我一次只会有 4 个被主动处理。

我现在的代码发送 4 个项目，然后等到所有 4 个完成，然后发送接下来的 4 个。我希望它发送前 4 个，然后当一个 observable 完成时，另一个进来待处理。

from([1, 2, 3, 4, 5, 6, 7, 8])
.pipe(
    bufferCount(4),
    concatMap(values => (
        from(values)
        .pipe(
            mergeMap(value => (
                timer(1000) // Async stuff would happen here
                .pipe(
                    mapTo(value),
                )
            )),
        )
    )),
)
.subscribe(console.log)

Answer 1

您必须使用mergeMap 的第二个参数，这是一个允许指定您想要的并发 级别的数字。

所以你的代码可能看起来像

from([1, 2, 3, 4, 5, 6, 7, 8])
.pipe(
    map(val => of(val)),
    mergeMap(val => val, 4),
)
.subscribe(console.log)

或

of([1, 2, 3, 4, 5, 6, 7, 8])
.pipe(
    mergeMap(val => val, 4),
)
.subscribe(console.log)

考虑concatMap 是mergeMap 并发级别设置为1。

阅读this SO post了解更多关于mergeMap的详细信息。