javascript - 使用Rxjs来并行调用具有延迟的同时请求块？

我有一个任务队列(长度为20)，其中每个任务都是要调用的ajax请求。

我想要 :

1)创建5个块(20/5 = 4个块)
2)执行每个块，其中每个块中的项将以1000 ms的延迟执行。
3)每个块项目完成后，等待3秒钟。

因此:

1..1sec(↦绿色).. 2..1sec((绿色).. 3..1sec(↦绿色).. 4..1sec(↦绿色).. 5 ......... ............ 3秒..........
6..1sec(↦绿).. 7..1sec(↦绿).. 8..1sec(9绿).. 9..1sec(↦绿).. 10 ......... ............ 3秒..........
...
11..1sec(↦绿色).. 12..1sec(↦绿色).. 13..1sec(↦绿色).. 14..1sec(↦绿色).. 15 ......... ............ 3秒..........
16..1sec(↦绿色).. 17..1sec(↦绿色).. 18..1sec(↦绿色).. 19..1sec(↦绿色).. 20

我确实做了一些接近的事情:



与:

from(this.httpCodes)
      .pipe(bufferCount(5),
       concatMap((i, y) => from(i).pipe(mergeMap(f => {
                                    this.httpCodes[f.index].wasExecuted = true;
                                     return this.ajaxAlike(f.data).pipe(
                                                               catchError(() => { return of(-1) }),
                                                               map((r) => ({ res: r, data: f }))
                                                                      )
                                                      })
                                        ,delay(3000) )),

      )

但是它没有按我的预期执行。我看不到块中每个项目之间的延迟

问题:

为什么我看到这么多请求，以及如何更改代码，以使块中的每个项目都延迟1秒执行(每秒应显示绿色)，并且-在每个块之后，等待3秒？

Online Demo

最佳答案

delay 运算符可延迟发射的项目。您似乎希望它会发射该项目，然后在发射下一个项目之前先“睡眠” 3秒钟。为此，您可以连接一个空的可观察到的延迟。

您可以创建以下可传递的 sleep 运算符:

const sleep = ms => concat(Rx.Observable.empty().pipe(delay(ms)))

并按如下所示使用它:

const {concatMap, concat, delay, bufferCount} = Rx.operators;

const sleep = ms => concat(Rx.Observable.empty().pipe(delay(ms)));

const ajaxAlike = call => Rx.Observable.of(call).pipe(delay(500));

Rx.Observable.range(0, 20).pipe(
  bufferCount(5),
  concatMap(calls =>
    Rx.Observable.from(calls).pipe(
      concatMap(call => ajaxAlike(call).pipe(sleep(1000))),
      sleep(3000)
    )
  )
)
.subscribe(console.log)

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.5.7/Rx.js"></script>