每 x ms 计算一个平均值

Question

我有一个发出一些数据的流。

const data$ = Rx
  .Observable
  .interval(500)
  .map(() => Math.random() * 100)

我想计算一段时间内所有产生值的平均值（例如每秒）

如果data$ 流发出-2-2-2-4|，我希望得到的流是---2---3|

您将如何使用 rxjs（第 5 版）来做到这一点

Answer 1

bufferTime 是一条路：

Rx.Observable
  .interval(500)
  .map(() => Math.random() * 100)
  .do(console.log)
  .bufferTime(2000)
  .map(items => {
    return items.reduce((acc, cur) => acc + cur, 0) / items.length;
  })
  .take(3)
  .subscribe(val => {
    console.log('mean: ' + val);
  })

<script src="https://npmcdn.com/@reactivex/rxjs@5.3.1/dist/global/Rx.js"></script>

好的，Maxime 的观点是正确的。如果您需要计算大量值的平均值，请考虑使用windowTime：

var interval = Rx.Observable.interval(2000);

Rx.Observable
  .interval(500)
  .map(() => Math.random() * 100)
  .do(console.log)
  .window(interval)
  .map(win => {
    return win.reduce((acc, cur) => {
      acc.count++;
      acc.sum += cur;
      return acc;
    }, {
      count: 0,
      sum: 0,
    });
  })
  .mergeAll()
  .filter(val => val.count != 0)
  .map(val => val.sum / val.count)
  .take(3)
  .subscribe(val => {
    console.log('mean: ' + val);
  })

<script src="https://npmcdn.com/@reactivex/rxjs@5.3.1/dist/global/Rx.js"></script>

Answer 2

如果您确实需要每秒捕获一个值，Sergey Sokolov 的回答可能没问题。但是，例如，宁愿每小时进行一次，bufferTime 可能必须在此期间将值保存在内存中。在这里，如果我理解您的问题，您就不需要保留这些值。

因此，这样做可能是一个更好的主意：

const { Observable } = Rx;

const tick$ = Observable.interval(2000);

const myObs$ = Observable
  .interval(500)
  .map(() => Math.random() * 100)
  .do(x => console.log(`Emit ${x}`));

const finalObs$ = tick$
  .withLatestFrom(myObs$)
  .map(([_, val]) => val)
  .do(x => console.warn(`Taken ${x}`));

// only to test
finalObs$.take(3).subscribe();

输出：

这里的每个console.log 都只是为了调试，但在您的最终观察中，您只能从上面的屏幕截图中获得黄色值。

这是一个有效的 Plunkr https://plnkr.co/edit/sR4Dg1gD3Zh2yWUtl0RI?p=preview