RXJS：如何以随机间隔（在指定范围内）生成数字流？

Question

我想使用 RXJS 设置一个 ORDERED 数据流，它以随机间隔（比如每 1-5 秒）发出一个数字，我想将其用作测试 RXJS 其他部分的时间随机数据源。以下代码以随机顺序生成项目（由于延迟），但我希望仅保留随机时间的顺序。

function randomDelay(bottom, top) {
  return Math.floor( Math.random() * ( 1 + top - bottom ) ) + bottom;
}

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

var subscription = source.subscribe(
  function (x) {
     $("#result").append('Next: ' +  JSON.stringify(x) +  '<br>');
  },
  function (err) {
     $("#result").append('Error: ' + err);
  },
  function () {
     $("#result").append('Completed');
  });

正在给我以下输出的变体：

Next: {"value":1,"interval":1229}
Next: {"value":2,"interval":321}
Next: {"value":4,"interval":645}
Next: {"value":5,"interval":28}
Next: {"value":9,"interval":728}
Next: {"value":10,"interval":269}
Next: {"value":3,"interval":107}
Next: {"value":6,"interval":265}
Next: {"value":8,"interval":1038}
Next: {"value":7,"interval":199}

Answer 1

我只是将这个问题作为我的基础，如果有人感兴趣，我必须将其更新到 RxJs 6。

const { range, of } = rxjs;
const { concatMap, delay } = rxjs.operators;

range(1, 10).pipe(
  concatMap(i => of(i).pipe(delay(1000 + (Math.random() * 4000))))
).subscribe(val => { console.log(val); });

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

Answer 2

使用concatMap 而不是flatMap。

这里的文档： https://github.com/Reactive-Extensions/RxJS/blob/master/doc/api/core/operators/concatmap.md

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

Answer 3

我建议将此问题分为两个子问题：i) 以随机间隔发射和 ii) 生成随机值。第一个问题可以通过合适的自定义 RxJS 观察者来解决（这就是 crazyObservable 的目的）；第二个问题可以通过自定义订阅 crazyObservable 来解决。

crazyObservable 有三个参数：totalNumberOfEmissions、timeInterval 和 totalNumberOfTimeIntervals；它返回一个自定义的 observable，在 totalNumberOfTimeIntervals 内随机发出 totalNumberOfEmissions。

要获得您想要的行为，请根据您的意愿设置 totalNumberOfEmissions、timeInterval 和 totalNumberOfTimeIntervals。这个Marble diagram 可以提供帮助。

var oneHourRandomTenHundredRequisitions$ = crazyObservable(1000,1000,60*60);

let testComponent1 = oneHourRandomTenHundredRequisitions$.subscribe({
next() { <a call to your random generator> } ,
error() { <your custom error call> },
complete() { <your custom complete call> }
});