RxJS操作符——数据流的重组

有时候获取到数据流之后，内部数据虽然都是需要的，但数据的形式或许不是我们想要的。这种时候就需要使用这些数据再去进一步加工得到想要的数据。这就是今天要谈到的这一类操作符了。

数据型重组

这一类重组操作符目的在于对数据流中的数据结果进行再处理。

reduce

类似于数组的reduce方法，为原数据流应用一个累加器函数，当元数据流完成时返回累加结果，同样可以指定一个初始值。

let source = Rx.Observable.of(1, 3, 5);
let target = source.reduce((acc, item) => acc + item);
// 1-3-5-|
// reduce
// ------9-|

average/sum/count

计算原数据流的平均值/总和/数量并在结束时发射。可以传入一个映射函数，则等价于先执行map函数，再进行操作符调用。

let source = Rx.Observable.of(1, 3, 5);
let target = source.average(item => item + 1);
// 1-3-5-|
// average
// ------4-|

max/min

获取原数据流的最大/最小值，可以传入一个重载的比较函数，和数组的sort方法类似。

let data = [
  {score: 1},
  {score: 2},
  {score: 3}
];
let source = Rx.Observable.from(data);
let target = source.min((d1, d2) => return d1.score > d2.score);
// --{score: 1}--{score: 2}--{score: 3}--|
// min
// ---------------------------{score: 1}-|

结构型重组

这一类重组操作符目的在于对数据流中的数据结构进行再处理。

buffer

该方法的参数为一个新的数据流，每当这个数据流生成一个元素，源数据流就将之前生成的数据打包植入一个数组。

let source = Rx.Observable.timer(0, 1000);
let bound = Rx.Observable.timer(2500);
let target = source.buffer(bound);
// 0---1---2---3---4---5---6---7---……
// buffer
// ----------[0，1，2]--[3，4，5]--[6, 7]……

window

该方法和buffer很相似，区别在于buffer是将元素转化为数组，而window是将元素再次转化为一个新的数据流。

参数和buffer相同是用来触发行为的数据流。返回的结果是一个数据流，但每个元素就是新的数据流。

let source = Rx.Observable.timer(0, 1000);
let bound = Rx.Observable.timer(2500);
let target = source.window(bound);
// 0---1---2---3---4---5--……
// bound
// ----------Ob{0, 1 ,2}---Ob{3, 4, 5}_

groupBy

该方法用于按指定键将源序列进行重组，每个分组重构成一个新的数据流。第一个参数是一个键选择函数，其参数是源数据流的元素，返回值将作为分组键的依据。也可以传入第2个类型为函数的参数先进行一次map映射再分组。

let source = Rx.Observable.timer(0, 1000);
let gf = item => item % 2 === 0 ? 'even' : 'odd';
let mf = item => String.fromCharCode(65 + item % 26);
let target = source.groupBy(gf, mf);
// 0---1---2---3---4---5---6--……
// groupBy
// -Ob{A, C, E……}--Ob{B, D, F……}--……

还有一些结构型重组的操作符都是基于以上三者衍生，在此就不做赘述。