既然选择了远方,便只顾风雨兼程.
该图片来自网络资源,若有侵权,请留言,我将自行删除
要说Android界网络请求框架,那就不得不提Rxjava + Retrofit.
Retrofit,是目前主流网络请求框架,功能强大,方便操作.
Rxjava,异步实现操作的库,可在线程间快速切换，并提供许多操作符,将复杂臃肿代码变得清晰易懂。
在Rxjava中通常表现为Observer(观察者)订阅Observable(可观察对象),通过创建可观察对象发送的数据流.在此过程中,可通过操作符,线程切换等操作最后由观察者接受并做出响应的一个过程.
Observable
Observable 具体表现为一个抽象类,实现ObservableSource接口
public abstract class Observable<T> implements ObservableSource<T> {
//......
}
ObservableSource 接口
public interface ObservableSource<T> {
/**
* Subscribes the given Observer to this ObservableSource instance.
* @param observer the Observer, not null
* @throws NullPointerException if {@code observer} is null
*/
void subscribe(@NonNull Observer<? super T> observer);
}
由此可见Observable是ObservableSource接口下的一个抽象类.可通过ObservableOnSubscribe 创建可观察对象发射数据流。
Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
emitter.onNext("hello world");
emitter.onComplete();
}
});
调用Observable.create方法，创建一个可观察对象，并通过onNext发送一条数据“hello world”，然后通过onComplete发送完成通知。
Observer
创建观察者,接收可观察对象发送的数据流
Observer<String> observer = new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(String s) {
LogUtils.e(s);
}
@Override
public void onError(Throwable e) {
e.printStackTrace();
}
@Override
public void onComplete() {
LogUtils.e("------接收完毕------");
}
};
关联观察者与可观察者对象
observable.subscribe(observer);
通过subscribe方法,观察者可订阅被观察者,一旦被订阅后,此时两者为一体.Observer便可对Observable中的行为作出响应。
控制台打印:
E/LogUtils: hello world
E/LogUtils: ------接收完毕------
Emitter / Observer
我们在使用Observable.create方法时,往参数中扔了一个ObservableOnSubscribe实例对象,而真正发射数据流是ObservableEmitter.
public interface ObservableEmitter<T> extends Emitter<T> {
//.......
}
public interface Emitter<T> {
/**
* 用来发送数据,可多次调用,每调用一次发送一次数据
* Signal a normal value.
* @param value the value to signal, not null
*/
void onNext(@NonNull T value);
/**
*用来发送异常通知，只发送一次，若多次调用只发送第一条
* Signal a Throwable exception.
* @param error the Throwable to signal, not null
*/
void onError(@NonNull Throwable error);
/**
* 用来发送完成通知，只发送一次，若多次调用只发送第一条
* Signal a completion.
*/
void onComplete();
}
- onNext() 用来发送数据流,可多次调用,每调用一次发送一次数据.
- onError() 发送异常通知,只调用一次.多次调用只发送第一条.
- onComplete() 发送完成数据流发送通知,只调用一次,多次调用只发送第一条.若数据在全部发送完之后均正常，可以调用onComplete发送一条完成通知
- 其中onError() 与 onComplete() 为唯一且互斥函数.
接口Observer中的三个方法（onNext,onError,onComplete）正好与Emitter中的三个方法相对应，对于Emitter中对应方法发送的数据或通知进行响应。
优化demo: 省略变量
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
emitter.onNext("hello world");
emitter.onComplete();
}
}).subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(String s) {
LogUtils.e(s);
}
@Override
public void onError(Throwable e) {
e.printStackTrace();
}
@Override
public void onComplete() {
LogUtils.e("------接收完毕------");
}
});
使用修饰符优化demo
Observable.just("Hello world").subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
LogUtils.e( s);
}
});
Observable.just("hello-world").subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception { // 接收正常发射的数据
LogUtils.e(s);
}
}, new Consumer<Throwable>() { //接收可观察者异常信息
@Override
public void accept(Throwable throwable) throws Exception {
throwable.printStackTrace();
}
}, new Action() { // 接收完成信息,作出响应行动。
@Override
public void run() throws Exception {
LogUtils.e("------接收完毕------");
}
});
- just 修饰符 : 发送单条数据.(数字,字符,字符串,数组,集合)都可当做单条数据发送.
- Consumer 可以看做是对观察者Observer功能单一化之后的产物.
第一个参数Consumer泛型为发射数据类型,通过函数accept进行数据流响应处理.
第二个参数Consumer规定泛型<Throwable>通过函数accept接收异常信息。
第三个参数Action也是对观察者Observer功能单一化之后的产物--行动，通过函数run接收完成信息，作出响应行动。
使用λ表达式
Observable.just("hello world").subscribe(s -> LogUtils.e(s));
发送数据序列
private void demo5(List<String> list) {
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
for (int i = 0; i < list.size(); i++) {
emitter.onNext(list.get(i) );
}
emitter.onComplete();
}
}).subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(String s) {
LogUtils.e(s);
}
@Override
public void onError(Throwable e) {
e.printStackTrace();
}
@Override
public void onComplete() {
LogUtils.e("------接收完毕------");
}
});
}
在subscribe方法中,遍历List 集合中的数据,通过 emitter.onNext函数将元素发射出去.遍历完成之后,由emitter.onComplete()函数告知发射数据完毕.
在Observer 接收数据中,函数onNext会一直接收emitter.onNext函数发射的数据,如遇到数据异常则函数onError会被回调.数据发射完毕,回调onComplete函数.
优化demo
private void demo6(List<String> list) {
Observable.fromIterable(list).subscribe(s -> LogUtils.e(s));
}
- 操作符 fromIterable:用来将一个可迭代对象中的元素逐一发送
Disposable
在Observer onSubscribe函数中参数是Disposable.用来切断observer和Observalbe之间的连接，当调用它的disposable()方法时，它就将观察者和被观察者之间的连接切断，从而使Observer收不到事件。确切的说是将Observer切断，不再接受来自被观察者的事件，但是被观察者事件仍在继续执行。
public interface Disposable {
/**
* Dispose the resource, the operation should be idempotent.
*/
void dispose();
/**
* Returns true if this resource has been disposed.
* @return true if this resource has been disposed
*/
boolean isDisposed();
}
- dispose 主动解除订阅.
- isDisposed 判断是否解除订阅.true为取消订阅,观察者不在接收数据流.
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
for (int i = 0; i < 10; i++) {
emitter.onNext(i );
LogUtils.e("发送者:" + i);
}
emitter.onComplete();
}
}).subscribe(new io.reactivex.Observer<Integer>() {
private Disposable mDisposable;
@Override
public void onSubscribe(Disposable d) {
this.mDisposable = d;
}
@Override
public void onNext(Integer s) {
if (s > 4){
mDisposable.dispose(); // 取消订阅
}
LogUtils.e("接收者: "+s);
}
@Override
public void onError(Throwable e) {
e.printStackTrace();
}
@Override
public void onComplete() {
LogUtils.e("结束---end");
}
});
}
log日志打印
2019-08-09 10:27:15.126 23763-23763/? E/LogUtils: 接收者: 0
2019-08-09 10:27:15.126 23763-23763/? E/LogUtils: 发送者-->:0
2019-08-09 10:27:15.126 23763-23763/? E/LogUtils: 接收者: 1
2019-08-09 10:27:15.126 23763-23763/? E/LogUtils: 发送者-->:1
2019-08-09 10:27:15.126 23763-23763/? E/LogUtils: 接收者: 2
2019-08-09 10:27:15.126 23763-23763/? E/LogUtils: 发送者-->:2
2019-08-09 10:27:15.127 23763-23763/? E/LogUtils: 接收者: 3
2019-08-09 10:27:15.127 23763-23763/? E/LogUtils: 发送者-->:3
2019-08-09 10:27:15.127 23763-23763/? E/LogUtils: 接收者: 4
2019-08-09 10:27:15.127 23763-23763/? E/LogUtils: 发送者-->:4
2019-08-09 10:27:15.127 23763-23763/? E/LogUtils: 接收者: 5
2019-08-09 10:27:15.127 23763-23763/? E/LogUtils: 发送者-->:5
2019-08-09 10:27:15.127 23763-23763/? E/LogUtils: 发送者-->:6
2019-08-09 10:27:15.127 23763-23763/? E/LogUtils: 发送者-->:7
2019-08-09 10:27:15.127 23763-23763/? E/LogUtils: 发送者-->:8
2019-08-09 10:27:15.127 23763-23763/? E/LogUtils: 发送者-->:9
在next函数中,当参数 s>4的情况下,mDisposable.dispose();取消订阅.停止接收Observable发送的事件，取消观察者和被观察者之间的联系后，Observable被观察者的事件仍在继续执行。
