SpringCloudRPC调用核心原理：RxJava响应式编程框架，聚合操作符

本节介绍RxJava的两个聚合型操作符&＃xff1a;count操作符和reduce操作符。

count操作符用来对源Observable流的数据项进行计数&＃xff0c;最后将总数弹射出来&＃xff1b;如果源流弹射错误&＃xff0c;就会将错误直接报出来&＃xff1b;在源Observable流没有终止前&＃xff0c;count操作符是不会弹射统计数据的。使用count操作符对数据流序列进行计数&＃xff0c;具体的执行流程如图4-9所示。

图4-9 使用count操作符对数据流序列进行计数

下面是一个使用count操作符的简单例子&＃xff0c;代码如下&＃xff1a;

package com.crazymaker.demo.rxJava.basic;
//省略import
&＃64;Slf4j
public class AggregateDemo
{/***演示count计数操作符*/&＃64;Testpublic void countDemo(){String[] items &＃61; {"one", "two", "three", "four"};Integer count &＃61; Observable.from(items).count() .toBlocking().single();log.info("计数的结果为 {}",count);}
}

运行以上代码&＃xff0c;输出的结果节选如下&＃xff1a;

[main] INFO 
c.c.d.r.basic.AggregateDemo - 计数的结果为 4可以看出&＃xff0c;count操作符将一个Observable源流转换成一个弹射单个值的Observable输出流&＃xff0c;输出流的唯一数据项的值为原始Observable流所弹射的数据项数量。

在上面的代码中&＃xff0c;为了获取count输出流中的数据项&＃xff0c;使用了toBlocking()和single()两个操作符。其中&＃xff0c;Observable.toBlocking()操作返回了一个BlockingObservable阻塞型实例&＃xff0c;该类型不是一种新的数据流&＃xff0c;仅仅是对源Observable的包装&＃xff0c;只是该类型会阻塞当前线程&＃xff0c;一直等待直到内部的源Observable弹射了自己想要的数据。BlockingObservable.single()方法表示阻塞当前线程&＃xff0c;直到从封装的源Observable获取到唯一的弹射数据元素项&＃xff0c;如果Observable源流弹射出的数据元素不止一个&＃xff0c;single()方法就会抛出异常。

Reduce&＃xff08;归约&＃xff09;操作符对一个Observable流序列的每一项应用一个归约函数&＃xff0c;最后将流的最终归约计算结果弹射出去。除了第一项之外&＃xff0c;reduce操作符会将上一个数据项应用归约函数的结果作为下一个数据项在应用归约函数时的输入。所以&＃xff0c;和scan操作符一样&＃xff0c;reduce操作符也有点类似递归操作。

假定归约函数为一个简单的累加函数&＃xff0c;然后使用reduce操作符对1~5的数据流序列进行归约&＃xff0c;其具体的归约流程如图4-10所示。

图4-10 reduce操作符对1~5的数据流序列的归约流程

使用reduce操作符实现对1~5的数据流序列的归约&＃xff0c;参考如下的实现代码&＃xff1a;

package com.crazymaker.demo.rxJava.basic;
//省略import
&＃64;Slf4j
public class AggregateDemo
{/**
演示
操作符 *演示reduce操作符*/&＃64;Testpublic void reduceDemo(){/***定义一个accumulator归约函数*/Func2 accumulator &＃61;new Func2(){&＃64;Overridepublic Integer call(Integer input1, Integer input2){log.info(" {} &＃43; {} &＃61; {} ", input1, input2, input1 &＃43; input2);return input1 &＃43; input2;}};/***使用reduce进行流归约*/Observable.range(1, 5).reduce(accumulator).subscribe(new Action1(){&＃64;Overridepublic void call(Integer sum){log.info(" 归约的结果: {} ", sum);}});}
}

运行以上代码&＃xff0c;输出的结果节选如下&＃xff1a;

[main] INFO c.c.d.r.basic.AggregateDemo - 1 &＃43; 2 &＃61; 3
[main] INFO c.c.d.r.basic.AggregateDemo - 3 &＃43; 3 &＃61; 6
[main] INFO c.c.d.r.basic.AggregateDemo - 6 &＃43; 4 &＃61; 10
[main] INFO c.c.d.r.basic.AggregateDemo - 10 &＃43; 5 &＃61; 15
[main] INFO c.c.d.r.basic.AggregateDemo - 归约的结果: 15

以上实例代码中&＃xff0c;reduce操作符对原始Observable流所弹射的第一项数据1应用归约函数&＃xff0c;得到中间结果1&＃xff1b;然后将第一个中间结果1连同原始流的第二项数据2一起填充给accumulator归约函数&＃xff0c;得到中间结果3。reduce持续对原始流进行迭代&＃xff0c;一直到原始流的最后一个数据项5&＃xff0c;reduce将5连同中间结果10一起填充给accumulator归约函数&＃xff0c;得到最终结果15。最后&＃xff0c;reduce会将最终结果15作为输出流的数据项弹射出去。reduce操作符与前面介绍的scan操作符很类似&＃xff0c;只是scan会弹出每次计算的中间结果&＃xff0c;而reduce只会弹出最后的结果。