concat操作符
concat操作符是將多個(gè)Observable 按傳入順序進(jìn)行輸出,O_a輸出完畢,O_b接著輸出:
Observable<Integer> range = Observable.range(1, 5);
List<Integer> data = Arrays.asList(6, 7, 8, 9, 10);
Observable<Integer> from = Observable.from(data);
Observable.concat(from,range).subscribe(new Observer<Integer>() {
@Override
public void onCompleted() {
LogUtils.d("----->onCompleted");
}
@Override
public void onError(Throwable e) {
LogUtils.d("----->onError");
}
@Override
public void onNext(Integer o) {
LogUtils.d("----->onNext:" + o);
}
});
運(yùn)行結(jié)果如下：
----->onNext:6
----->onNext:7
----->onNext:8
----->onNext:9
----->onNext:10
----->onNext:1
----->onNext:2
----->onNext:3
----->onNext:4
----->onNext:5
----->onCompleted
combineLastest操作符
combineLastest() 是將第一個(gè)Observable的【最新】數(shù)據(jù)與后面的Observable數(shù)據(jù)項(xiàng)按某種規(guī)則合并啤它。
Observable.combineLatest(range, from, new Func2<Integer, Integer, Integer>() {
@Override
public Integer call(Integer integer, Integer integer2) {
Log.d("rxandroidtest","----->i1:"+integer+" i2:"+integer2);
return integer + integer2;
}
}).subscribe(new Subscriber<Integer>() {
@Override
public void onCompleted() {
Log.d("rxandroidtest","----->onCompleted");
}
@Override
public void onError(Throwable e) {
Log.d("rxandroidtest","----->onError");
}
@Override
public void onNext(Integer o) {
Log.d("rxandroidtest","----->onNext:" + o);
}
});
運(yùn)行結(jié)果如下：
----->i1:5 i2:6
----->onNext:11
----->i1:5 i2:7
----->onNext:12
----->i1:5 i2:8
----->onNext:13
----->i1:5 i2:9
----->onNext:14
----->i1:5 i2:10
----->onNext:15
----->onCompleted
join操作符
在a的生命周期內(nèi)：b輸出的數(shù)據(jù)項(xiàng)與a輸出的數(shù)據(jù)項(xiàng)每個(gè)合并,直到b輸出下一項(xiàng)：
為什么要加延遲摩瞎？是因?yàn)榭赡躡的輸出會(huì)因?yàn)閍的生命周期結(jié)束了而b還未開(kāi)始
rang 和 from 參考上面
from.observeOn(Schedulers.newThread());
range.observeOn(Schedulers.newThread());
range.join(from, new Func1<Integer, Observable<Integer>>() {
@Override
public Observable<Integer> call(Integer integer) {
Log.d("join","left:"+integer);
return Observable.just(integer).delay(1, TimeUnit.SECONDS);
}
}, new Func1<Integer, Observable<Integer>>() {
@Override
public Observable<Integer> call(Integer integer) {
Log.d("join","right:"+integer);
return Observable.just(integer).delay(1, TimeUnit.SECONDS);
}
}, new Func2<Integer, Integer, Integer>() {
@Override
public Integer call(Integer integer, Integer integer2) {
Log.d("join","left:"+integer + " right:"+integer2);
return integer + integer2;
}
}).subscribe(new Subscriber<Integer>() {
@Override
public void onCompleted() {
Log.d("join","onCompleted！");
}
@Override
public void onError(Throwable e) {
Log.d("join","onError:"+e.getMessage());
}
@Override
public void onNext(Integer o) {
Log.d("join","onNext:"+o);
}
});
運(yùn)行結(jié)果如下：
D/join: left:1
D/join: left:2
D/join: left:3
D/join: left:4
D/join: left:5
D/join: right:6
D/join: left:5 right:6
D/join: onNext:11
D/join: left:2 right:6
D/join: onNext:8
D/join: left:1 right:6
D/join: onNext:7
D/join: left:4 right:6
D/join: onNext:10
D/join: left:3 right:6
D/join: onNext:9
D/join: right:7
D/join: left:5 right:7
D/join: onNext:12
D/join: left:2 right:7
D/join: onNext:9
D/join: left:1 right:7
D/join: onNext:8
D/join: left:4 right:7
D/join: onNext:11
D/join: left:3 right:7
D/join: onNext:10
D/join: right:8
D/join: left:5 right:8
D/join: onNext:13
D/join: left:2 right:8
D/join: onNext:10
D/join: left:1 right:8
D/join: onNext:9
D/join: left:4 right:8
D/join: onNext:12
D/join: left:3 right:8
D/join: onNext:11
D/join: right:9
D/join: left:5 right:9
D/join: onNext:14
D/join: left:2 right:9
D/join: onNext:11
D/join: left:1 right:9
D/join: onNext:10
D/join: left:4 right:9
D/join: onNext:13
D/join: left:3 right:9
D/join: onNext:12
D/join: right:10
D/join: left:5 right:10
D/join: onNext:15
D/join: left:2 right:10
D/join: onNext:12
D/join: left:1 right:10
D/join: onNext:11
D/join: left:4 right:10
D/join: onNext:14
D/join: left:3 right:10
D/join: onNext:13
D/join: onCompleted！
merge 操作符
按照兩個(gè)Observable提交結(jié)果的時(shí)間順序進(jìn)行輸出
Observable.merge(range, from).subscribe(new Subscriber<Integer>() {
@Override
public void onCompleted() {
Log.d("join","onCompleted捅位！");
}
@Override
public void onError(Throwable e) {
Log.d("join","onError:"+e.getMessage());
}
@Override
public void onNext(Integer integer) {
Log.d("join","onNext:"+integer);
}
});
運(yùn)行結(jié)果如下：
onNext:1
D/join: onNext:2
D/join: onNext:3
D/join: onNext:4
D/join: onNext:5
D/join: onNext:6
D/join: onNext:7
D/join: onNext:8
D/join: onNext:9
D/join: onNext:10
D/join: onCompleted！
例子2
Observable<Long> interval_1 = Observable.interval(3, TimeUnit.SECONDS).map(aLong -> aLong * 1).take(5);
Observable<Long> interval_100 = Observable.interval(3, TimeUnit.SECONDS).map(aLong -> aLong * 100).take(5);
Observable.merge(interval_1,interval_100).subscribe(new Subscriber<Long>() {
@Override
public void onCompleted() {
Log.d("join","onCompleted蚀乔！");
}
@Override
public void onError(Throwable e) {
Log.d("join","onError:"+e.getMessage());
}
@Override
public void onNext(Long aLong) {
Log.d("join","onNext:"+aLong);
}
});
運(yùn)行結(jié)果如下：
D/join: onNext:0
D/join: onNext:0
D/join: onNext:1
D/join: onNext:100
D/join: onNext:2
D/join: onNext:200
D/join: onNext:3
D/join: onNext:300
D/join: onNext:4
D/join: onNext:400
D/join: onCompleted持寄！
mergeDelayError操作符
在merge過(guò)程中 一旦某一個(gè)Observable中出現(xiàn)錯(cuò)誤,就會(huì)停止合 并,mergeDelayError操作符會(huì)把錯(cuò)誤放在所有結(jié)果都和并完成后才執(zhí)行:
Observable<Long> interval_1 = Observable.interval(3, TimeUnit.SECONDS).map(aLong -> {
if(aLong > 2) throw new RuntimeException();
return aLong * 1;
}).take(5);
Observable<Long> interval_100 = Observable.interval(3, TimeUnit.SECONDS).map(aLong -> aLong * 100).take(5);
Observable.mergeDelayError(interval_1,interval_100).subscribe(new Subscriber<Long>() {
@Override
public void onCompleted() {
Log.d("join","onCompleted崔步！");
}
@Override
public void onError(Throwable e) {
Log.d("join","onError:"+ e);
}
@Override
public void onNext(Long aLong) {
Log.d("join","onNext:"+aLong);
}
});
運(yùn)行結(jié)果如下：
onNext:0
onNext:0
onNext:1
onNext:100
onNext:2
onNext:200
onNext:300
onNext:400
onError:java.lang.RuntimeException
startWith操作符
在源Observable輸出之前插入指定數(shù)據(jù)項(xiàng):
range.startWith(8,9).subscribe(new Subscriber<Integer>() {
@Override
public void onCompleted() {
Log.d("join","onCompleted稳吮！");
}
@Override
public void onError(Throwable e) {
Log.d("join","onError:"+ e);
}
@Override
public void onNext(Integer integer) {
Log.d("join","onNext:"+integer);
}
});
運(yùn)行結(jié)果如下：
onNext:8
onNext:9
onNext:1
onNext:2
onNext:3
onNext:4
onNext:5
onCompleted！
switchOnNext操作符
把一組Observable轉(zhuǎn)換成一個(gè)Observable, 這組Observable中的每一個(gè)Observable所產(chǎn)生的結(jié)果,如果在同一個(gè)時(shí)間段內(nèi)兩個(gè)或者多個(gè)Observable提交結(jié)果,只取最后一個(gè)Observable提交的結(jié)果
Observable<Observable<Long>> longObservable = Observable.interval(10, TimeUnit.SECONDS).map(new Func1<Long, Observable<Long>>() {
@Override
public Observable<Long> call(final Long aLong1) {
return Observable.interval(1, TimeUnit.SECONDS).map(new Func1<Long, Long>() {
@Override
public Long call(Long aLong) {
// Log.d("join","------>aLong1:" + aLong);
// Log.d("join","------>aLong:" + aLong);
return aLong + (aLong1*100);
}
}).take(3);
}
}).take(3);
Observable.switchOnNext(longObservable).subscribe(new Observer<Long>() {
@Override
public void onCompleted() {
Log.d("join","------>onCompleted()");
}
@Override
public void onError(Throwable e) {
Log.d("join","------>onError()" + e);
}
@Override
public void onNext(Long integer) {
Log.d("join","------>onNext()" + integer);
}
});
運(yùn)行結(jié)果如下：
onNext()0
onNext()1
onNext()2
onNext()100
onNext()101
onNext()102
onNext()200
onNext()201
onNext()202
onCompleted()
如果同一時(shí)間有多個(gè)Observable發(fā)射井濒，只取最后一個(gè)
Observable<Observable<Long>> longObservable = Observable.interval(3, TimeUnit.SECONDS).map(new Func1<Long, Observable<Long>>() {
@Override
public Observable<Long> call(final Long aLong1) {
return Observable.interval(1, TimeUnit.SECONDS).map(new Func1<Long, Long>() {
@Override
public Long call(Long aLong) {
// Log.d("join","------>aLong1:" + aLong);
// Log.d("join","------>aLong:" + aLong);
return aLong + (aLong1*100);
}
}).take(3);
}
}).take(3);
Observable.switchOnNext(longObservable).subscribe(new Observer<Long>() {
@Override
public void onCompleted() {
Log.d("join","------>onCompleted()");
}
@Override
public void onError(Throwable e) {
Log.d("join","------>onError()" + e);
}
@Override
public void onNext(Long integer) {
Log.d("join","------>onNext()" + integer);
}
});
運(yùn)行結(jié)果如下：
少了2灶似，102
onNext()0
onNext()1
onNext()100
onNext()101
onNext()200
onNext()201
onNext()202
onCompleted()
zip 操作符
是將同索引位置的數(shù)據(jù)項(xiàng)進(jìn)行合并,放單的數(shù)據(jù)項(xiàng)丟棄
Observable.zip(Observable.just(1, 2, 3), Observable.just(10, 20, 30), new Func2<Integer, Integer, Integer>() {
@Override
public Integer call(Integer integer, Integer integer2) {
return integer + integer2;
}
}).subscribe(
new Observer<Integer>() {
@Override
public void onCompleted() {
Log.d("join","------>onCompleted()");
}
@Override
public void onError(Throwable e) {
Log.d("join","------>onError()" + e);
}
@Override
public void onNext(Integer integer) {
Log.d("join","------>onNext()" + integer);
}
}
);
運(yùn)行結(jié)果如下：
onNext()11
onNext()22
onNext()33
onCompleted()
