转换操作符
doOnNext()
其实觉得 doOnNext 应该不算一个操作符，但考虑到其常用性，我们还是咬咬牙将它放在了这里。它的作用是让订阅者在接收到数据之前干点有意思的事情。假如我们在获取到数据之前想先保存一下它，我们就可以这样实现。
Observable.just(1,2,5,8).filter(new Predicate<Integer>() {
@Override
public boolean test(@NonNull Integer integer) throws Exception {
return integer>0;
}
}).subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.d(TAG, "accept："+integer);
}
});
filter()
filter是一个过滤操作符，根据自己想过滤的数据加入相应的逻辑判断，返回true则表示数据满足条件，返回false则表示数据被过滤。最后过滤出的数据发送给下游。
Observable.just(1,2,3,-5,4,-2,5,8).filter(new Predicate<Integer>() {
@Override
public boolean test(@NonNull Integer integer) throws Exception {
return integer>0;
}
}).subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.d(TAG, "accept："+integer);
}
});
map()
map()把上游发送的每一个事件都应用一个函数, 每个事件都按照指定的函数去变化。也就是你可以通过一个函数对发送过来的数据进行预处理，得到想要的数据再发送给观察者。
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
Log.d(TAG, "Observable发出：1");
e.onNext(1);//向下游（观察者）发射内容1
Log.d(TAG, "Observable发出：2");
e.onNext(2);
Log.d(TAG, "Observable发出：3");
e.onNext(3);
}
}).map(new Function<Integer, String>() {
@Override
public String apply(@NonNull Integer integer) throws Exception {
Integer num=integer*10;
return "结果被map改变了："+num;
}
}).subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
Log.d(TAG, "accept："+s);
}
});
flatMap()
上游每发送一个事件, flatMap都将创建一个新的Observable, 然后发送转换之后的新的事件, 下游接收到的就是这些新的Observable发送的数据，但flatMap并不保证事件的顺序。
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
Log.d(TAG, "Observable发出：1");
e.onNext(1);//向下游（观察者）发射内容1
Log.d(TAG, "Observable发出：2");
e.onNext(2);
Log.d(TAG, "Observable发出：3");
e.onNext(3);
}
}).flatMap(new Function<Integer, ObservableSource<?>>() {
@Override
public ObservableSource<?> apply(@NonNull Integer integer) throws Exception {
final List<String> list = new ArrayList<>();
for (int i = 0; i < 3; i++) {
list.add("fromIterable转换结果：" + integer);
}
return Observable.fromIterable(list).delay(10, TimeUnit.MILLISECONDS);
}
}).subscribe(new Consumer<Object>() {
@Override
public void accept(Object o) throws Exception {
Log.d(TAG,""+o.toString());
}
});
concatMap()
concatMap和flatMap一样，但是保证事件顺序
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
Log.d(TAG, "Observable发出：1");
e.onNext(1);//向下游（观察者）发射内容1
Log.d(TAG, "Observable发出：2");
e.onNext(2);
Log.d(TAG, "Observable发出：3");
e.onNext(3);
}
}).concatMap(new Function<Integer, ObservableSource<?>>() {
@Override
public ObservableSource<?> apply(@NonNull Integer integer) throws Exception {
final List<String> list = new ArrayList<>();
for (int i = 0; i < 3; i++) {
list.add("fromIterable转换结果：" + integer);
}
return Observable.fromIterable(list).delay(10, TimeUnit.MILLISECONDS);
}
}).subscribe(new Consumer<Object>() {
@Override
public void accept(Object o) throws Exception {
Log.d(TAG,""+o.toString());
}
});
zip()
Zip通过一个函数将多个Observable发送的事件结合到一起，然后发送这些组合到一起的事件. 它按照严格的顺序应用这个函数。它只发射与发射数据项最少的那个Observable一样多的数据。
Observable<Integer> observable1 = Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
Log.d(TAG, "observable1：A1");
e.onNext(1);//向下游（观察者）发射内容1
Thread.sleep(1000);
Log.d(TAG, "observable1：A2");
e.onNext(2);
Thread.sleep(1000);
Log.d(TAG, "observable1：A3");
e.onNext(3);
e.onComplete();
}
}).subscribeOn(Schedulers.io());//使用新线程，同时发射数据，不使用则按顺序发射
Observable<Integer> observable2 = Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
Log.d(TAG, "observable1：B1");
e.onNext(1);//向下游（观察者）发射内容1
Thread.sleep(1000);
Log.d(TAG, "observable1：B2");
e.onNext(2);
Thread.sleep(1000);
Log.d(TAG, "observable1：B3");
e.onNext(3);
//Thread.sleep(1000);//这里不能sleep，因为父线程可能已经结束（observable1已结束，zip）
Log.d(TAG, "observable1：B4");
e.onNext(4);
Log.d(TAG, "observable1：B5");
e.onNext(5);
e.onComplete();
}
}).subscribeOn(Schedulers.io());
Observable.zip(observable1, observable2, new BiFunction<Integer, Integer, Object>() {
@Override
public Object apply(@NonNull Integer integer, @NonNull Integer integer2) throws
Exception {
return integer + integer2;
}
}).subscribe(new Observer<Object>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
Log.d(TAG,"onSubscribe");
}
@Override
public void onNext(@NonNull Object o) {
Log.d(TAG,"onNext:"+o.toString());
}
@Override
public void onError(@NonNull Throwable e) {
Log.d(TAG,"onError");
}
@Override
public void onComplete() {
Log.d(TAG,"onComplete");
}
});
更多操作符可参考官方文档：
http://reactivex.io/documentation/operators.html
