1.Map操作符
作用：将被观察者发送的事件转换为任意类型的事件
例子：数据类型的转换
/**
* map 操作符
* 事件转换
* 将被观察者发送的事件转换为任意的类型事件。
* 例子：数据类型转换
*/
private void map() {
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
e.onNext(111);
}
//使用map操作符进行事件转换
}).map(new Function<Integer, String>() {
@Override
public String apply(Integer integer) throws Exception {
return String.valueOf(integer) + "你好";
}
//被观察者订阅事件
}).subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
Log.e(TAG, "转换结果： " + s);
}
});
}
输出结果:
image.png
2.FlatMap操作符
作用：将被观察者发送的多个事件进行拆分，分别转换为不同的事件
，然后合并成新的事件
例子：将每个Integer数据分别转换成一个List，然后再迭代
/**
*FlatMap 操作符
* 将被观察者的事件顺序拆分
*
* 单独转换
* 在合并成新的事件
*
*/
private void flatMap(){
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
e.onNext(1);
e.onNext(2);
e.onNext(3);
Log.e(TAG, "----- " );
}
}).flatMap(new Function<Integer, ObservableSource<String>>() {
@Override
public ObservableSource<String> apply(Integer integer) throws Exception {
List<String> list = new ArrayList<>();
//分别对被观察者的事件进行加工转换i
for (int i = 0; i <2 ; i++) {
list.add("我是事件"+integer+"拆分后的子事件为"+i);
}
Log.e(TAG, "---------");
//拆分后合并
//
return Observable.fromIterable(list);
}
}).subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
Log.e(TAG, "accept: "+s );
}
});
}
输出结果:
image.png
3.ConcatMap操作符
和FlatMap操作符原理一样，区别在于 FlatMap把事件拆分再合并的时候，顺序可能和被观察者发送的事件顺序不一样，而ConcatMap操作符处理后的事件顺序和被观察者发送事件的顺序是一样的。
/**
* 类似FlatMap（）操作符
* 但是新事件循序还是和观察者的顺序一样
*/
private void concatMap(){
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
e.onNext(1);
e.onNext(2);
e.onNext(3);
e.onNext(4);
e.onNext(5);
e.onNext(6);
Log.e(TAG, "----- " );
}
}).concatMap(new Function<Integer, ObservableSource<String>>() {
@Override
public ObservableSource<String> apply(Integer integer) throws Exception {
List<String> list = new ArrayList<>();
for (int i = 0; i <2 ; i++) {
list.add("我是事件"+integer+"拆分后的子事件为"+i);
}
Log.e(TAG, "----- " );
return Observable.fromIterable(list);
}
}).subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
Log.e(TAG, "accept: "+s );
}
});
}
输出结果：
image.png
4.Buffer操作符
把被观察者的事件进行缓存，然后每次从缓存中拿出一定数量事件发送给观察者。
/**
* buffer 操作符
* 定期从 被观察者（Obervable）需要发送的事件中
* 获取一定数量的事件 & 放到缓存区中，最终发送
* 缓存被观察者的发送事件
*
*/
private void buffer(){
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
e.onNext(1);
e.onNext(2);
e.onNext(3);
e.onNext(4);
}
}).buffer(2,1)//参数1：每次发送给观察者事件的数量。参数2：每次从缓存中取出事件的数量
.subscribe(new Consumer<List<Integer>>() {
@Override
public void accept(List<Integer> i) throws Exception {
for (int j = 0; j <i.size() ; j++) {
Log.e(TAG, "accept: "+i.get(j) );
}
}
});
}
输出结果
image.png
