RxJS:如何设置exhaustMap并发？

Question

我们可以在 mergeMap 上使用并发，但我认为它导致了累积。
我可以为exhaustMap 设置并发吗？

interval(500).pipe(
  mergeMap(timer(3000), 3)
)

Answer 1

您可以自己创建这样的运算符。这是一个可能的实现:

const { interval, timer, of, empty } = rxjs
const { mergeMap, map, tap, filter } = rxjs.operators

const exhaustMapWithConcurrency = 
  (selector, concurrency = 1) => 
    source => {
  let active = 0;
  return source.pipe(
    filter(_ => active < concurrency),
    tap(_ => active++),
    mergeMap(selector),
    tap(_ => active--)
  )
}

interval(500).pipe(
  exhaustMapWithConcurrency(x => 
    timer(3000).pipe(
      map(_ => x)
    ), 3)
)
.subscribe(console.log)

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.5.2/rxjs.umd.min.js"></script>