Rxjava 操作符-4-组合操作符
2019-01-03 本文已影响0人
揭穿套路
1、startWith
map 操作符的作用是对 对象进行转换
Observable.range(0, 5).filter(new Predicate<Integer>() {
@Override
public boolean test(Integer integer) throws Exception {
return integer > 2;//自定义的条件,只有符合条件的结果才会提交给观察者
}
}).subscribe(consumer);
12-29 12:00:37.769 461-461/com.rxjavademo E/tag: 3
12-29 12:00:37.774 461-461/com.rxjavademo E/tag: 4
2、merge
merge 合并请求，但是不能保证顺序
Observable<Integer> firstObservable = Observable.just(0, 1, 2).subscribeOn(Schedulers.io());
Observable<Integer> secondObservable = Observable.just(3, 4, 5);
Observable.merge(firstObservable, secondObservable).subscribe(consumer, onErrorAction);
12-29 13:28:12.280 6062-6062/com.rxjavademo E/tag: 3
12-29 13:28:12.285 6062-6062/com.rxjavademo E/tag: 4
12-29 13:28:12.290 6062-6062/com.rxjavademo E/tag: 5
12-29 13:28:12.297 6062-6062/com.rxjavademo E/tag: 0
12-29 13:28:12.303 6062-6062/com.rxjavademo E/tag: 1
12-29 13:28:12.307 6062-6062/com.rxjavademo E/tag: 2
merge 只要有一个请求出错了就会中断 其他请求，
mergeDelayError 延迟错误，请求都走完 再报错
3、concat
merge 合并请求，严格按照顺序执行
Observable<Integer> firstObservable = Observable.just(0, 1, 2).subscribeOn(Schedulers.io());
Observable<Integer> secondObservable = Observable.just(3, 4, 5);
Observable.concat(firstObservable, secondObservable).subscribe(consumer, onErrorAction);
12-29 13:32:13.606 6062-6062/com.rxjavademo E/tag: 0
12-29 13:32:13.612 6062-6062/com.rxjavademo E/tag: 1
12-29 13:32:13.615 6062-6062/com.rxjavademo E/tag: 2
12-29 13:32:13.621 6062-6062/com.rxjavademo E/tag: 3
12-29 13:32:13.624 6062-6062/com.rxjavademo E/tag: 4
12-29 13:32:13.627 6062-6062/com.rxjavademo E/tag: 5
concat 只要有一个请求出错了就会中断 其他请求，
concatArrayDelayError 延迟错误，请求都走完 再报错
Observable.concat(
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
emitter.onError(new NullPointerException()); // 发送Error事件，因为无使用concatDelayError，所以第2个Observable将不会发送事件
emitter.onComplete();
}
}),
Observable.just(4, 5, 6))
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Integer value) {
Log.e("tag", "onNext"+ value );
}
@Override
public void onError(Throwable e) {
Log.e("tag"," onError");
}
@Override
public void onComplete() {
Log.e("tag","onComplete");
}
});
12-29 13:45:04.863 13255-13255/com.rxjavademo E/tag: onNext1
12-29 13:45:04.866 13255-13255/com.rxjavademo E/tag: onNext2
12-29 13:45:04.863 13255-13255/com.rxjavademo E/tag: onNext3
12-29 13:45:04.866 13255-13255/com.rxjavademo E/tag: onError
4、zip
zip操作符，会将多个Observable对象转换成一个Observable对象然后进行发送，zip 跟merge的区别是 zip 等待多个请求全部成功以后再一起通知
//
Observable<Integer> integerObservable = Observable.range(0, 4);
Observable<String> stringObservable = Observable.just("a", "b", "c", "d");
Observable.zip(integerObservable, stringObservable, new BiFunction<Integer, String, Object>() {
@Override
public Object apply(Integer integer, String s) throws Exception {
return "数字为:" + integer + "……字符为：" + s;
}
}).subscribe(consumer);
12-29 13:47:43.590 13255-13255/com.rxjavademo E/tag: 数字为:0……字符为：a
12-29 13:47:43.595 13255-13255/com.rxjavademo E/tag: 数字为:1……字符为：b
12-29 13:47:43.601 13255-13255/com.rxjavademo E/tag: 数字为:2……字符为：c
12-29 13:47:43.607 13255-13255/com.rxjavademo E/tag: 数字为:3……字符为：d
5、combineLatest
不同于zip操作符的是，combineLatest会将 其中一个Observable发射数据后 取其最后一项，组合所有未发送的Observable
integerObservable = Observable.just(1, 2, 3);
stringObservable = Observable.just("a", "b", "c");
Observable.combineLatest(integerObservable, stringObservable, new BiFunction<Integer, String, Object>() {
@Override
public Object apply(Integer num, String info) {
//在这里的转换关系为将数字与字串内容进行拼接
return "数字为:" + num + "……字符为：" + info;
}
}).subscribe(consumer);
12-29 13:55:57.154 13255-13255/com.rxjavademo E/tag: 数字为:3……字符为：a
12-29 13:55:57.163 13255-13255/com.rxjavademo E/tag: 数字为:3……字符为：b
12-29 13:55:57.169 13255-13255/com.rxjavademo E/tag: 数字为:3……字符为：c
integerObservable = Observable.just(1, 2, 3).delay(200, TimeUnit.MILLISECONDS, AndroidSchedulers.mainThread());
stringObservable = Observable.just("a", "b", "c");
Observable.combineLatest(integerObservable, stringObservable, new BiFunction<Integer, String, Object>() {
@Override
public Object apply(Integer num, String info) {
//在这里的转换关系为将数字与字串内容进行拼接
return "数字为:" + num + "……字符为：" + info;
}
}).subscribe(consumer);
12-29 14:23:26.859 15193-15193/com.rxjavademo E/tag: 数字为:1……字符为：c
12-29 14:23:26.869 15193-15193/com.rxjavademo E/tag: 数字为:2……字符为：c
12-29 14:23:26.875 15193-15193/com.rxjavademo E/tag: 数字为:3……字符为：c
6、ignoreElements
ignoreElements操作符忽略所有源Observable产生的结果，只会执行onCpmpleted()或者onError()方法
Observable.range(0, 5).ignoreElements().subscribe(new Action() {
@Override
public void run() throws Exception {
Log.e("tag","onCompleted()" );
}
}, consumer);
12-29 12:39:59.976 3700-3700/com.rxjavademo E/tag: onCompleted()
7、throttleFirst
throttleFirst操作符，会定期发送这个时间段里源Observable发送的第一个数据
Observable
.create(new ObservableOnSubscribe<Object>() {
@Override
public void subscribe(ObservableEmitter<Object> emitter) throws Exception {
for (int i = 0; i < 10; i++) {
emitter.onNext(i);
//线程休眠100毫秒
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
})
.throttleFirst(200, TimeUnit.MILLISECONDS)
.subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(consumer);
12-29 13:55:57.154 13255-13255/com.rxjavademo E/tag: 数字为:3……字符为：a
12-29 13:55:57.163 13255-13255/com.rxjavademo E/tag: 数字为:3……字符为：b
12-29 13:55:57.169 13255-13255/com.rxjavademo E/tag: 数字为:3……字符为：c
6、throttleWithTimeout
如果两次数据的发射间隔小于指定时间，就会丢弃前一次的数据,直到指定时间内都没有新数据发射时才进行发射
Observable
.create(new ObservableOnSubscribe<Object>() {
@Override
public void subscribe(ObservableEmitter<Object> emitter) throws Exception {
emitter.onNext(1);
try {
Thread.sleep(500);
} catch (InterruptedException e) {
throw Exceptions.propagate(e);
}
emitter.onNext(2);
try {
Thread.sleep(500);
} catch (InterruptedException e) {
throw Exceptions.propagate(e);
}
emitter.onNext(3);
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
throw Exceptions.propagate(e);
}
emitter.onNext(4);
emitter.onNext(5);
emitter.onNext(6);
emitter.onComplete();
}
})
.throttleWithTimeout(800, TimeUnit.MILLISECONDS)
.subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(consumer);
12-29 13:10:12.382 4834-4834/com.rxjavademo E/tag: 3
12-29 13:10:12.582 4834-4834/com.rxjavademo E/tag: 6
