我想使用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}
使用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();
[我只是将此问题作为另一个问题的基础,如果有人感兴趣,则必须将其更新为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>
您想看看https://github.com/cszredwan/crazyObservable吗?我相信这会有所帮助,这是一个可观察到的习惯,可以在固定的时间范围内发出(随机!)固定数量的数据。