RxJava系列教程:
? Creating Observables(Observable的创建操作符) 【视频教程】
? Transforming Observables(Observable的转换操作符) 【视频教程】
? Filtering Observables(Observable的过滤操作符) 【视频教程】
? Combining Observables(Observable的组合操作符) 【视频教程】
? Error Handling Operators(Observable的错误处理操作符) 【视频教程】
? Observable Utility Operators(Observable的辅助性操作符) 【视频教程】
? Conditional and Boolean Operators(Observable的条件和布尔操作符) 【视频教程】
? Mathematical and Aggregate Operators(Observable数学运算及聚合操作符) 【视频教程】
? 其他如observable.toList()、observable.connect()、observable.publish()等等; 【视频教程】
3. RxJava Observer与Subcriber的关系 【视频教程】
4. RxJava线程控制(Scheduler) 【视频教程】
5. RxJava 并发之数据流发射太快如何办(背压(Backpressure)) 【视频教程】
All
All操作符根据一个函数对源Observable发射的所有数据进行判断,最终返回的结果就是这个判断结果。这个函数使用发射的数据作为参数,内部判断所有的数据是否满足我们定义好的判断条件,如果全部都满足则返回true,否则就返回false。
示例代码:
Observable.from(studentList).all(new Func1<Student, Boolean>() {
@Override
public Boolean call(Student student) {
return student.age > 25;
}
}).subscribe(new Action1<Boolean>() {
@Override
public void call(Boolean value) {
System.out.println("onSuccess value = " + value);
}
});
List<Student> studentList = new ArrayList<Student>() {
{
add(new Student("Stay", 28));
add(new Student("谷歌小弟", 23));
add(new Student("Star", 25));
}
};
输出:
onSuccess value = false
Amb
传递两个或多个(至多9个)Observable给Amb时,它只发射其中首先发射数据或通知(onError或onCompleted)的那个Observable的所有数据,而其他所有的Observable的发射物将被丢弃。
有一个类似的对象方法ambWith。Observable.amb(o1,o2)和o1.ambWith(o2)是等价的。
示例代码:
Observable.amb(Observable.just(1,2,3,4,5).delay(2, TimeUnit.SECONDS),
Observable.just(100,101))
.subscribe(new Action1<Integer>() {
@Override
public void call(Integer value) {
System.out.println("call value = " + value);
}
});
输出:
call value = 100
call value = 101
Contains、IsEmpty 、exists
Contains操作符用来判断源Observable所发射的数据是否包含某一个数据,如果包含会返回true,如果源Observable已经结束了却还没有发射这个数据则返回false。
IsEmpty:相关的一个操作符IsEmpty用于判定原始Observable是否没有发射任何数据。
exists:RxJava中还有一个exists操作符,它通过一个谓词函数测试原始Observable发射的数据,只要任何一项满足条件就返回一个发射true的Observable,否则返回一个发射false的Observable。
示例代码:
//Contains:判定一个Observable是否发射一个特定的值
Observable.just(4,5,6)
.contains(4)
.subscribe(new Action1<Boolean>() {
@Override
public void call(Boolean aBoolean) {
System.out.println("contains(4):"+aBoolean);
}
});
//isEmpty:判定原始Observable是否没有发射任何数据
Observable.just(4,5,6)
.isEmpty()
.subscribe(new Action1<Boolean>() {
@Override
public void call(Boolean aBoolean) {
System.out.println("isEmpty():"+aBoolean));
}
});
//exists操作符,它通过一个谓词函数测试原始Observable发射的数据,
// 只要任何一项满足条件就返回一个发射true的Observable,
// 否则返回一个发射false的Observable。
Observable.just(4,5,6)
.exists(new Func1<Integer, Boolean>() {
@Override
public Boolean call(Integer integer) {
return integer<5;
}
})
.subscribe(new Action1<Boolean>() {
@Override
public void call(Boolean aBoolean) {
System.out.println("exists():"+aBoolean);
}
});
输出:
contains(4):true
isEmpty():false
exists():true
DefaultIfEmpty
??DefaultIfEmpty简单的精确地发射原始Observable的值,如果原始Observable没有发射任何数据正常终止(以onCompleted的形式),DefaultIfEmpty返回的Observable就发射一个你提供的默认值。
示例代码:
Observable.empty()
.defaultIfEmpty(10)
.subscribe(new Action1<Integer>() {
@Override
public void call(Integer value) {
System.out.println("defaultIfEmpty:" + value);
}
});
输出:
defaultIfEmpty:10
SequenceEqual
判定两个Observables是否发射相同的数据序列。 传递两个Observable给SequenceEqual操作符,它会比较两个Observable的发射物,如果两个序列是相同的(相同的数据,相同的顺序,相同的终止状态),它就发射true,否则发射false。
示例代码:
Observable.sequenceEqual(
//第一个Observable延迟1秒发射数据
Observable.just(4,5,6).delay(1,TimeUnit.SECONDS),
Observable.just(4,5,6))
.subscribe(new Action1<Boolean>() {
@Override
public void call(Boolean aBoolean) {
System.out.println("sequenceEqual:"+aBoolean);
}
});
.subscribe(aBoolean -> Log.v(TAG, "sequenceEqual:"+aBoolean));
输出:
sequenceEqual:true
SkipUntil
SkipUntil订阅原始的Observable,但是忽略它的发射物,直到第二个Observable发射了一项数据那一刻,它开始发射原始Observable。
示例代码:
Observable.interval(1, TimeUnit.SECONDS) //从0开始
.take(6)
.skipUntil(Observable.just(10).delay(3,TimeUnit.SECONDS))
.subscribe(new Subscriber<Long>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Long value) {
System.out.println("onNext value = " + value);
}
});
输出:
skipUntil value = 3
skipUntil value = 4
skipUntil value = 5
SkipWhile
SkipWhile订阅原始的Observable,但是忽略它的发射物,直到你指定的某个条件变为false的那一刻,它开始发射原始Observable。
示例代码:
Observable.interval(1, TimeUnit.SECONDS) //从0开始
.take(6)
.skipWhile(new Func1<Long, Boolean>() {
@Override
public Boolean call(Long aLong) {
return aLong<3; //舍弃原Observable发射的数据,直到发射的数据>=3,才继续发射
}
})
.subscribe(new Subscriber<Long>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Long value) {
System.out.println("skipWhile value = " + value);
}
});
输出:
skipWhile value = 3
skipWhile value = 4
skipWhile value = 5
TakeUntil
??takeUntil与skipUntil操作符作用相反,订阅并开始发射原始Observable,它还监视你提供的第二个Observable。如果第二个Observable发射了一项数据或者发射了一个终止通知( onError通知或一个onCompleted通知),TakeUntil返回的Observable会停止发射原始Observable并终止。
示例代码:
Observable.interval(500, TimeUnit.MILLISECONDS) //从0开始
.takeUntil(Observable.just(100, 101, 102).delay(3, TimeUnit.SECONDS))
.subscribe(new Subscriber<Long>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Long value) {
System.out.println("takeUntil value = " + value);
}
});
输出:
takeUntil value = 0
takeUntil value = 1
takeUntil value = 2
takeUntil value = 3
takeUntil value = 4
TakeWhile
takeWhile与skipWhile操作符作用相反,发射原始Observable,直到你指定的某个条件不成立的那一刻,它停止发射原始Observable,并终止自己的Observable。
示例代码:
Observable.interval(1, TimeUnit.SECONDS) //从0开始
.takeWhile(new Func1<Long, Boolean>() {
@Override
public Boolean call(Long aLong) {
return aLong<3;
}
})
.subscribe(new Subscriber<Long>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Long value) {
System.out.println("takeWhile value = " + value);
}
});
输出:
takeWhile value = 0
takeWhile value = 1
takeWhile value = 2