首先，将您的异步操作移出subscribe，它不适合异步操作。
您可以使用mergeMap（别名flatMap）或concatMap。 （我提到了他们两个，但concatMap 实际上是mergeMap，concurrent 参数设置为 1。）设置不同的并发参数很有用，因为有时您希望限制并发查询的数量，但是仍然运行几个并发。
source.concatMap(item => {
if (item == 'do-something-async-and-wait-for-completion') {
return Rx.Observable.timer(5000)
.mapTo(item)
.do(e => console.log('okay, we can continue'));
} else {
// do something synchronously and keep on going immediately
return Rx.Observable.of(item)
.do(e => console.log('ready to go!!!'));
}
}).subscribe();
我还将展示如何对通话进行速率限制。 忠告： 仅在您实际需要时限制速率，例如在调用每秒或每分钟只允许一定数量请求的外部 API 时。否则最好只限制并发操作的数量，让系统以最大速度移动。
我们从下面的sn-p开始：
const concurrent;
const delay;
source.mergeMap(item =>
selector(item, delay)
, concurrent)
接下来，我们需要为concurrent、delay 选择值并实现selector。 concurrent 和 delay 密切相关。例如，如果我们想每秒运行 10 个项目，我们可以使用 concurrent = 10 和 delay = 1000（毫秒），也可以使用 concurrent = 5 和 delay = 500 或 concurrent = 4 和 delay = 400。每秒的项目数将始终为concurrent / (delay / 1000)。
现在让我们实现selector。我们有几个选择。我们可以为selector 设置一个最小的执行时间，我们可以给它添加一个恒定的延迟，我们可以在结果可用时立即发出结果，我们只能在最小延迟过去后发出结果等等。它是甚至可以使用 timeout 运算符添加超时。方便。
设置最短时间，尽早发送结果：
function selector(item, delay) {
return Rx.Observable.of(item)
.delay(1000) // replace this with your actual call.
.merge(Rx.Observable.timer(delay).ignoreElements())
}
设置最短时间，延迟发送结果：
function selector(item, delay) {
return Rx.Observable.of(item)
.delay(1000) // replace this with your actual call.
.zip(Rx.Observable.timer(delay), (item, _))
}
添加时间，提前发送结果：
function selector(item, delay) {
return Rx.Observable.of(item)
.delay(1000) // replace this with your actual call.
.concat(Rx.Observable.timer(delay).ignoreElements())
}
添加时间，延迟发送结果：
function selector(item, delay) {
return Rx.Observable.of(item)
.delay(1000) // replace this with your actual call.
.delay(delay)
}