javascript - 如何在 RXJS 中实现带背压的轮询

Question

我想要轮询一个响应速度较慢的 HTTP API，因此我不想同时多次调用此 API。

我想做的一个例子是:

const interval = Rx.Observable.interval(250).take(5); // Poll every 250ms

function simulateMaybeSlowHttpCall() {
  return interval.delay(500).take(1); // The service takes 500ms to answer
}

 interval
   .mergeMap(val =>simulateMaybeSlowHttpCall().map(x => val), 1) // max concurrent is 1
   .subscribe(val => console.log(val));

这里会显示这段代码12345

但我不想打无用的电话。上面的代码运行 250*5 = 1250 毫秒，1 次调用需要 500 毫秒，所以我想显示:

135

所以我的问题是:当将并发设置为1(或任何其他值)时，如何丢弃所有未立即完成的调用？

JsFiddle:https://jsfiddle.net/zra3zxhs/63/

Answer 1

使用并发度为 1 的 mergeMap 相当于 concatMap。事实上，就是这样concatMap is implemented 。这就是示例中的每个间隔都会影响 HTTP 请求的原因:它们已排队。

如果您希望避免在 HTTP 请求处于待处理状态时启动或排队，您可以使用 exhaustMap :

interval
  .exhaustMap(val => simulateMaybeSlowHttpCall().map(x => val))
  .subscribe(val => console.log(val));

当使用 exhaustMap 时，它收到的任何 next 通知都会被忽略，直到内部可观察对象(HTTP 请求)完成为止。