深入浅出RxJava_05[转换操作&数学运算]

本教程基于RxJava1.x版本进行全面讲解，后续课程将陆续更新,敬请关注…

在观察者和被观察者之间，有些传输的数据需要进行转换之后才能用，比如：有时候拿到的是学校的某个班级学生的名单,但是我们需要知道他们的语文成绩排列。这样就是需要将ArrayList转换成ArrayList队列。

下列提供了一系列的数据切换操作符：

Buffer - 将发送的多个数据变成发送n个队列的数据,每个队列的最大长度为buffer()函数指定的参数大小。

Window - 定期将来自原始Observable的数据分解为一个Observable窗口.

Map - 将某个发送的数据转换成另外一个数据

flatMap - 实际上就是将发送的一个类型的数据转换成另外一个数据类型的对象

GroupBy - 根据指定的key将相同类型的发送数据存入到n个子Observable中。

Scan - 对所有的发送数据进行某个函数运行，将第n和第n+1项进行计算，计算后的结果再和n+2项进行运行，一次类推。

1.Buffer

比如下面的数据，分别发送三个字符串

Observable.just("Hello Android !", "Hello Java", "Hello C");

如果我们在该被观察者后面添加buffer操作符，并指定缓存2项，代码如下：

final Observable<String> observable = Observable.just("Hello Android !", "Hello Java", "Hello C");Observable<List<String>> bufferObservable = observable.buffer(2);

那么它会第一次发送2个List数据，分别是：

List<String>(){"Hello Android !", "Hello Java"}
List<String>(){"Hello C"}

下面的完整示例代码如下：

//这里模拟正常发送三个单一值的被观察者
final Observable<String> observable = Observable.just("Hello Android !", "Hello Java", "Hello C");//定义一个缓存的被观察者 每次缓存2个 缓存的数据自上面observable对象获取
Observable<List<String>> bufferObservable = observable.buffer(2);
//订阅对象并获取缓存被观察者发送出来的数据
bufferObservable.subscribe(new Action1<List<String>>() {
@Override
public void call(List<String> strings) {
Log.i(TAG, "call:--------");
for (int i = 0; i < strings.size(); i++) {
Log.i(TAG, "call: "+strings.get(i));
}
}
});

输出

call:--------
call: Hello Android !
call: Hello Java
call:--------
call: Hello C

buffer()实际上就是将发送的n个数据转换成发送x个队列

2.Window

定期将来自原始Observable的数据分解为一个Observable窗口，发射这些窗口，而不是每次发射一项数据

Window和Buffer类似，但不是发射来自原始Observable的数据包，它发射的是Observables，这些Observables中的每一个都发射原始Observable数据的一个子集，最后发射一个onCompleted通知。

Observable.just(1, 2, 3, 4, 5)
//将一个Observable发射出去的数据分解为多个Observable对象 每个Observable发射2个数据
.window(2)
.subscribe(new Action1<Observable<Integer>>() {
@Override
public void call(Observable<Integer> innerObservable) {
Log.i(TAG, "call: ---------");
innerObservable.subscribe(new Action1<Integer>() {
@Override
public void call(Integer value) {
Log.i(TAG, "call: "+value);
}
});
}
});

输出：

call: ---------
call: 1
call: 2
call: ---------
call: 3
call: 4
call: ---------
call: 5

3.Map

学校新生入住。在数据录入的时候，发现张三的名字录入出错，应该改为张三丰。现在 我们要重新为他录入信息。代码如下：

private ArrayList initPersons() {
ArrayList<Student> persons = new ArrayList<>();
persons.add(new Student("张三", 16));
persons.add(new Student("李四", 17));
persons.add(new Student("王二麻子", 18));
return persons;
}

ArrayList<Student> students = initPersons();
for (int i = 0; i < students.size(); i++) {
Student student = students.get(i);
if (student.name.equals("张三")){
student.name="张三丰";
}
}

RxJava这样转换：

Observable.from(initPersons())
//将张三的名字改为张三丰
.map(new Func1<Student, Student>() {
@Override
public Student call(Student student) {
//循环检查每个学生的名字，如果找到张三 则改为张三丰
if (student.name.equals("张三")){
student.name="张三丰";
}
return student;
}
})
.subscribe(new Action1<Student>() {
@Override
public void call(Student student) {
Log.i(TAG, "call: "+student);
}
});

map()实际上就是将某个发送的数据转换成另外一个数据

4.FlatMap

今天学校组建了体操队。学校又招了几个学生 代码如下：

private ArrayList<Student> initStudents() {
ArrayList<Student> persons = new ArrayList<>();
persons.add(new Student("张三", 11));
persons.add(new Student("李四", 12));
persons.add(new Student("王二麻子", 13));
persons.add(new Student("李雷", 19));
persons.add(new Student("韩梅梅", 18));
persons.add(new Student("韩红", 17));
return persons;
}

上级要求，学体操要从娃娃抓起，让我们找出年龄小于15岁的学生，组建一个体操队，并且给小学生创建一个特定的小学生类。

public static class LittleStudent extends Student{

public LittleStudent(String name, int age) {
super(name, age);
}
}

于是代码如下：

ArrayList<Student> students = initStudents();

//封装小学生集合
ArrayList<LittleStudent> littleStudents=new ArrayList<>();
for (int i = 0; i < students.size(); i++) {
Student student = students.get(i);
if (student.age<15){
littleStudents.add(new LittleStudent(student.name,student.age));
}
}

RxJava这样转换：

Observable.from(initStudents())
.flatMap(new Func1<Student, Observable<LittleStudent>>() {
@Override
public Observable<LittleStudent> call(Student student) {
if (student.age<15){
return Observable.just(new LittleStudent(student.name,student.age));
}
return null;
}
})
.subscribe(new Action1<LittleStudent>() {
@Override
public void call(LittleStudent littleStudent) {
Log.i(TAG, "call: "+littleStudent);
}
});

flatMap()实际上就是将发送的一个类型的数据转换成另外一个数据类型的对象

5.GroupBy

将一个Observable分拆为一些Observables集合，它们中的每一个发射原始Observable的一个子序列。哪个数据项由哪一个Observable发射是由groupBy内部的函数判定的，这个函数给每一项指定一个Key，Key相同的数据会被同一个Observable发射。它返回Observable的一个特殊子类GroupedObservable，实现了GroupedObservable接口的对象有一个额外的方法getKey，这个Key用于将数据分组到指定的Observable。

//模拟学校录入的一系列的同名学生信息
private Observable<Person> getPersons(){
return Observable.from(new Person[]{
new Person("zhangsan",18),
new Person("zhangsan",20),
new Person("lisi",19),
new Person("lisi",33),
new Person("wangwu",20),
new Person("wangwu",22),
new Person("wangwu",21)
});
}

实现归类调用

final Observable<GroupedObservable<String, Person>> observable = getPersons()
//根据用户的名称来归类
.groupBy(new Func1<Person, String>() {
@Override
public String call(Person person) {
return person.name;
}
});
observable.subscribe(new Action1<GroupedObservable<String, Person>>() {
//每归一类 则调用一次该方法
@Override
public void call(GroupedObservable<String, Person> observablelist) {
Log.i(TAG, "call: ----"+observablelist.getKey());
//打印队列中的每个元素
observablelist.subscribe(new Action1<Person>() {
@Override
public void call(Person person) {
Log.i(TAG, "call: "+person.name+"  "+person.age);
}
});

}
});

根据指定的key将相同类型的发送数据存入到n个子Observable中

6.Scan

Scan操作符对原始Observable发射的第一项数据应用一个函数，然后将那个函数的结果作为自己的第一项数据发射。它将函数的结果同第二项数据一起填充给这个函数来产生它自己的第二项数据。它持续进行这个过程来产生剩余的数据序列。这个操作符在某些情况下被叫做accumulator。

比如我们想从1加到5，思路是这样的：

1+2=3;
3+3=6;
6+4=10;
10+5=15;

RxJava这样转换：

//实现从1加到5的总数
Observable.just(1, 2, 3, 4, 5)
.scan(new Func2<Integer, Integer, Integer>() {
//定义每个子项的合并规则
@Override
public Integer call(Integer sum, Integer item) {
return sum + item;
}
})
.subscribe(new Action1<Integer>() {
@Override
public void call(Integer result) {
Log.i(TAG, "call: " + result);
}
});

输出

Next: 1
Next: 3
Next: 6
Next: 10
Next: 15
Sequence complete.

对所有的发送数据进行某个函数运行，将第n和第n+1项进行计算，计算后的结果再和n+2项进行运行，一次类推

在开发的过程，不可避免的会使用某些常见的数学计算，比如计算一个队列的平均数/求和/最大最小值/获取个数等等

关于数学的计算，这里RxJava提供了一个依赖包。其下载的地址是RxJavaMath

其开发包提供了一个核心帮助类：MathObservable

7.Average

求某个队列的平均数，这个队列可以是int类型 也可以是double类型等。以下例子计算1到6的平均数

Observable<Integer> o1 =
MathObservable.averageInteger(Observable.just(1,2,3,4,5,6));
o1.subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.i(TAG, "call: "+integer);
}
});

8.Max/Min

Min操作符操作一个发射数值的Observable并发射单个值：最小的那个值。

Max操作符操作一个发射数值的Observable并发射单个值：最大的那个值。

以下例子列出1到6的最小值：

Observable<Integer> o1 =
MathObservable.min(Observable.just(1,2,3,4,5,6));

o1.subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.i(TAG, "call: "+integer);
}
});

以下例子列出1到6的最大值：

Observable<Integer> o1 =
MathObservable.max(Observable.just(1,2,3,4,5,6));

o1.subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.i(TAG, "call: "+integer);
}
});

9.Count

count函数主要列出发送队列的个数。

Observable<Integer> o1 =Observable.just(1,2,3,4,5,4);
Observable<Integer> count = o1.count();

count.subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
//打印出6
Log.i(TAG, "call: "+integer);
}
});

10.Sum

计算Observable发射的数值的和并发射这个和.

RxJava的实现是sumDouble, sumFloat, sumInteger, sumLong，它们不是RxJava核心模块的一部分，属于rxjava-math模块。你可以使用一个函数，计算Observable每一项数据的函数返回值的和。

Observable<Integer> o1 =
MathObservable.sumInteger(Observable.just(1,2,3,4,5,4));

o1.subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.i(TAG, "call: "+integer);
}
});