RxJS 5.5.10：如何节流基于Promise的Observable？

首先咱们来拆解你遇到的问题：

为什么会报错？ #

你看到的TypeError: this.durationSelector is not a function，一方面是因为你把throttle用在了错误的位置，另一方面，对于固定时长的节流需求，你应该用throttleTime而非throttle（throttle更适合动态调整等待时间的场景）。

更关键的核心逻辑问题：Rx.Observable.fromPromise(mongo.AllWishes)只会发射一次值——就是整个数组，在这个流上用节流操作符完全没用，因为没有后续的值需要限制。

正确的实现代码 #

你需要先把数组拆分成逐个发射单个元素的Observable，再在这个流上应用节流，才能实现每5秒输出一个url的效果：

Rx.Observable.fromPromise(mongo.AllWishes)
  .flatMap(array => Rx.Observable.from(array)) // 先把数组拆成单个元素的流
  .throttleTime(5000) // 在这里添加节流：每5秒允许一个值通过
  .pluck('url')
  .filter(s => s !== undefined)
  .subscribe(m => console.log(m))

如果你需要用throttle（动态场景） #

如果未来你要根据每个元素的值动态设置节流时间，可以用throttle，但同样要放在拆分后的流上：

Rx.Observable.fromPromise(mongo.AllWishes)
  .flatMap(array => Rx.Observable.from(array))
  .throttle(() => Rx.Observable.interval(5000)) // 返回Observable指定节流时长
  .pluck('url')
  .filter(s => s !== undefined)
  .subscribe(m => console.log(m))

总结 #

• 节流操作符要放在单个元素发射的流之后，而非整个数组发射的流之前
• 固定时长节流用throttleTime(毫秒数)更简洁，throttle适合动态调整时长的场景

内容的提问来源于stack exchange，提问作者CptNapalm