如何在 rxjs 上每 1 秒产生价值

Question

如果我已经有一个 observable，那么我应该使用什么运算符来制作它 可以观察到每 1 秒产生价值吗？

// this just an example, In my project, I can't control when the 
// observable will produce value. I can assume that it will produce
// value very fast.
const obs = from([1,2,3,4,5]);

obs 将很快发出值 1,2,3...。但是如果我想要它怎么办 每1秒发出一次值？我的意思是，只要确保 obs 不会发出值 快吗？

我从reactivex 中检查了文档，但找不到操作员这样做。 例如 delay ，它只是让值的产生延迟了一些时间，但值之间的相对时间间隔被保留，而 debounceTime 确实会定期产生价值，但会忽略那个时间窗口的价值。

有人能告诉我如何让可观察的在期间产生价值吗 并且不会遗漏或忽略值？

Answer 1

您可以像这样使用 interval 可观察到的 zip 它：

import { zip, from, interval } from rxjs

const obs = zip(
  from([1,2,3,4,5]),
  interval(1000),
  (val, i) => val // Just emit the value
)

obs.subscribe(val => console.log(val))

如果您希望立即发出第一个值，那么您可以使用 timer 而不是 interval：

import { zip, from, timer } from rxjs

const obs = zip(
  from([1,2,3,4,5]),
  timer(0, 1000),
  (val, i) => val // Just emit the value
)

obs.subscribe(val => console.log(val))

如果您愿意，也可以使用管道，如下所示：

import { from, interval } from rxjs
import { zip } from rxjs/operators

const obs = from([1,2,3,4,5])
  .pipe(
    zip(interval(1000), val => val)
  )

obs.subscribe(val => console.log(val))

Answer 2

您确实可以使用delay，但您必须将每个延迟值转换为 Observable，然后将它们与concatMap 连接到序列中。

from([1,2,3,4,5]).pipe(
  concatMap(v => of(v).pipe(delay(1000))),
)

您可以使用zip，但这仅适用于测试数据。 zip 仅在所有源 Observable 发出相同数量的项目时才会发出。这意味着如果一个 Observable 发射得非常快，然后放慢速度，然后第二个 Observable 开始快速发射，它将比 1s 延迟发射得更快。即使interval 和zip 组合在一起也会发生这种情况，因为zip 在内部缓冲所有值，因此它可能会堆叠来自interval 的多个发射，然后如果第二个 Observable 发射非常快，则立即重新发射所有这些值。

Answer 3

你可以试试这种方法

var source = Rx.Observable
  .range(1, 10)
  .concatMap(function (x) {
    return Rx.Observable
      .of(x)
      .delay(1000);
  })
 .timeInterval();