1. RxJava2 : 什么是观察者模式
2. RxJava2 : 创建操作符(无关时间)
3. Rxjava2 : 创建操作符(有关时间)
4. Rxjava2 : 变换操作符
5. Rxjava2 : 判断操作符
6. Rxjava2 : 筛选操作符
7. Rxjava2 : 合并操作符
8. Rxjava2 : do操作符
9. Rxjava2 : error处理
10. Rxjava2 : 重试
11. Rxjava2 : 线程切换
基本结构
Observable.subscribe(Observer);
Observable 的创建(无关时间)
| api | use |
|---|---|
| create | {{Observable.create}} |
| just | {{Observable.just}} |
| fromArray / fromIterable | {{Observable.fromArray}} |
| range / rangeLong | {{Observable.range}} |
| empty / error / never | {{empty/error/never }} |
| defer | {{Observable.defer}} |
需要注意的是:
1.在使用操作符的时候,一定要清楚操作的是元素
2.以上表格中创建方式均不涉及时间,会按照顺序发送,但几乎是同一时刻全部发送
Observable.create
- ObservableEmitter : 发射器
- e.onNext(value) : 被 Observer 的 onNext() 接收
e.onError(Throwable e) : 被 Observer 的 onError(Throwable e) 接收
e.onComplete() : 被 Observer 的 onComplete() 接收
其中 : e.onError(Throwable e) 和 e.onComplete() 均有可能作为本次发送的终止 - 发送完毕后,不会自动发送onComplete,需手动调用终止
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
e.onNext(1);
e.onNext(2);
e.onNext(3);
//e.onError(Throwable throwable);
e.onComplete();
}
}).subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Integer integer) {
Log.d(TAG, "integer:" + integer);
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "onError");
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete");
}
});
Observable.just
- 快速发送,但不能发送超过10个元素
- 并不局限于发送同一种类型,可以混合发送
- 发送完毕后,会自动发送 onComplete
Observable.just(1,"one",2,"two")
.subscribe(new Observer<Serializable>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Serializable serializable) {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
Observable.fromArray / Observable.fromIterable
Observable.fromArray:
1.支持传入数组,也支持采用不定参数
2.此时, 数组中的每一项元素均作为rxjava中的一个元素,而并不是整个数组作为rxjava当中的元素
3.发送完毕后,会自动发送 onCompleteObservable.fromIterable:
1.均于Observable.fromArray 类似, 只不过支持传入的是集合
Integer[] integers = { 1,2,3 };
Observable.fromArray(integers).subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Integer integer) {
Log.d(TAG, "integer:" + integer);
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "onError");
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete");
}
});
log
02-12 14:10:46.273 26133-26133/... D/SplashActivity: integer:1
02-12 14:10:46.273 26133-26133/... D/SplashActivity: integer:2
02-12 14:10:46.273 26133-26133/... D/SplashActivity: integer:3
02-12 14:10:46.283 26133-26133/... D/SplashActivity: onComplete
Observable.range
- range
1.设定一个初始值,发送固定的数目
2.range为Integer类型
3.第一个参数为起始值,第二个参数为需要发送的个数
4.发送完毕后,会自动发送 onComplete - rangeLong
1.均和 range 相同,不过发送的为Long类型
Observable.range(3,10)
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Integer integer) {
Log.d(TAG, "integer:" + integer);
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "onError");
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete");
}
});
log
02-12 14:44:24.793 27930-27930/... D/SplashActivity: integer:3
02-12 14:44:24.793 27930-27930/... D/SplashActivity: integer:4
02-12 14:44:24.793 27930-27930/... D/SplashActivity: integer:5
02-12 14:44:24.793 27930-27930/... D/SplashActivity: integer:6
02-12 14:44:24.793 27930-27930/... D/SplashActivity: integer:7
02-12 14:44:24.793 27930-27930/... D/SplashActivity: integer:8
02-12 14:44:24.793 27930-27930/... D/SplashActivity: integer:9
02-12 14:44:24.793 27930-27930/... D/SplashActivity: integer:10
02-12 14:44:24.793 27930-27930/... D/SplashActivity: integer:11
02-12 14:44:24.793 27930-27930/... D/SplashActivity: integer:12
02-12 14:44:24.793 27930-27930/... D/SplashActivity: onComplete
empty / error / never
- empty
直接发送onComplete - error (Throwable throwable)
直接发送onError - never
什么都不会发送
Observable.empty()
.subscribe(new Observer<Object>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Object o) {
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "onError");
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete");
}
});
log
02-12 14:20:42.793 26705-26705/... D/SplashActivity: onComplete
Observable.defer
- 获取发送值是在subscribe的时候
- 使用场景并不多,了解即可
- 发送完毕后,会自动发送 onComplete
//打印出的i的值为15
private Integer i = 10;
Observable<Integer> observable = Observable
.defer(new Callable<ObservableSource<? extends Integer>>() {
@Override
public ObservableSource<? extends Integer> call() throws Exception {
return Observable.just(i);
}
});
i = 15;
observable.subscribe(i -> {
Log.d(TAG, "i" + i);
});
//打印出的i的值为10
private Integer i = 10;
Observable<Integer> observable = Observable.just(i);
i = 15;
observable.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Integer integer) {
Log.d(TAG, "integer:" + integer);
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "onError");
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete");
}
});
