在 RxJS 5 中我会这样做:
Observable.range(1, 25)
.bufferCount(5)
.concatMap(batch => { // process images
console.log('process', batch);
return Observable.from(batch)
.mergeMap(val => Observable.of('p' + val).delay(300))
.toArray();
})
.concatMap(batch => { // send images
console.log('send batch', batch);
return Observable.from(batch)
.mergeMap(val => Observable.of('s' + val).delay(500))
.toArray();
})
.subscribe(val => {
// console.log('response');
console.log('response', val);
});
With bufferCount http://reactivex.io/rxjs/class/es6/Observable.js~Observable.html#instance-method-bufferCount运算符我将输入数组分成 5 个项目的批次。然后每批首先处理第一个concatMap()
(我故意使用 concat 因为我想等到嵌套的 Observable 完成)。然后将处理后的数据发送到另一个concatMap()
将其发送到您的服务器。
我用的是两个delay()
操作员模拟不同的任务花费不同的时间。在我们的例子中,处理图像非常快,所以第一个concatMap
将比第二个更快地发射物品concatMap
能够将它们发送到服务器,这没问题。处理后的图像将被堆叠在里面concatMap
并将陆续分批发送。
该演示的输出如下所示:
process [ 1, 2, 3, 4, 5 ]
send batch [ 'p1', 'p2', 'p3', 'p4', 'p5' ]
process [ 6, 7, 8, 9, 10 ]
process [ 11, 12, 13, 14, 15 ]
response [ 'sp1', 'sp2', 'sp3', 'sp4', 'sp5' ]
send batch [ 'p6', 'p7', 'p8', 'p9', 'p10' ]
process [ 16, 17, 18, 19, 20 ]
process [ 21, 22, 23, 24, 25 ]
response [ 'sp6', 'sp7', 'sp8', 'sp9', 'sp10' ]
send batch [ 'p11', 'p12', 'p13', 'p14', 'p15' ]
response [ 'sp11', 'sp12', 'sp13', 'sp14', 'sp15' ]
send batch [ 'p16', 'p17', 'p18', 'p19', 'p20' ]
response [ 'sp16', 'sp17', 'sp18', 'sp19', 'sp20' ]
send batch [ 'p21', 'p22', 'p23', 'p24', 'p25' ]
response [ 'sp21', 'sp22', 'sp23', 'sp24', 'sp25' ]
观看现场演示:https://jsbin.com/mileqa/edit?js,控制台 https://jsbin.com/mileqa/edit?js,console
但是,如果您希望始终首先处理一个批次而不是发送它,并且当发送它时继续处理另一个批次,您必须将第二个内部 Observable 从concatMap
在结束时toArray()
在第一个concatMap()
call.
.concatMap(batch => { // process images
console.log('process', batch);
return Observable.from(batch)
.mergeMap(val => Observable.of('p' + val).delay(100))
.toArray()
.concatMap(batch => { // send images
console.log('send batch', batch);
return Observable.from(batch)
.mergeMap(val => Observable.of('s' + val).delay(500))
.toArray();
});
})
观看现场演示:https://jsbin.com/sabena/2/edit?js,控制台 https://jsbin.com/sabena/2/edit?js,console
这会产生如下输出:
process [ 1, 2, 3, 4, 5 ]
send batch [ 'p1', 'p2', 'p3', 'p4', 'p5' ]
response [ 'sp1', 'sp2', 'sp3', 'sp4', 'sp5' ]
process [ 6, 7, 8, 9, 10 ]
send batch [ 'p6', 'p7', 'p8', 'p9', 'p10' ]
response [ 'sp6', 'sp7', 'sp8', 'sp9', 'sp10' ]
process [ 11, 12, 13, 14, 15 ]
send batch [ 'p11', 'p12', 'p13', 'p14', 'p15' ]
response [ 'sp11', 'sp12', 'sp13', 'sp14', 'sp15' ]
process [ 16, 17, 18, 19, 20 ]
send batch [ 'p16', 'p17', 'p18', 'p19', 'p20' ]
response [ 'sp16', 'sp17', 'sp18', 'sp19', 'sp20' ]
process [ 21, 22, 23, 24, 25 ]
send batch [ 'p21', 'p22', 'p23', 'p24', 'p25' ]
response [ 'sp21', 'sp22', 'sp23', 'sp24', 'sp25' ]
可以看到“处理”、“发送批次”和“响应”日志是按顺序排列的。
RxJS 4 中的实现应该几乎相同(只是运算符名称可能略有不同)。
在 RxJS 4 中还有controlled()操作员 https://github.com/Reactive-Extensions/RxJS/blob/master/doc/gettingstarted/backpressure.md#controlled-observablesRxJS 5 中不存在(还?)。我做的事情也许与你需要的非常相似。