1、前言
- 在Android中，常见的异步操作类有
Handler、AsyncTask、HandlerThread、IntentService等； - 而
RxJava凭借着自己简洁清晰的链式风格，强大的操作符集合，已经成为开发者爱不释手的异步操作类；
2、简介
RxJava实质上是一个符合响应式编程的异步请求框架，可以把它看作一个观察者模式的扩展，即存在观察者、被观察者、订阅功能；并且RxJava的链式调用风格，让代码层次简洁清晰，越复杂的逻辑越利于RxJava的发挥；RxJava可以自由的切换上下游的线程环境，开发者可以更灵活的实现异步操作；
3、使用步骤
3.1、添加依赖
implementation 'io.reactivex.rxjava2:rxjava:2.1.4'
implementation 'io.reactivex.rxjava2:rxandroid:2.0.2'
3.2、创建被观察者
Observable observable = Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
emitter.onNext(4);
emitter.onComplete();
}
});
由emitter发射器调用onNext()发送事件，发送完事件后调用onComplete()；
3.3、创建被观察者
Observer observer = new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Integer integer) {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
};
当订阅后，观察者最先回调onSubscribe()，随后回调onNext()消费事件，当消息队列中没有消息就回调onComplete()结束传递，假如事件异常就会回调onError()；
3.4、订阅
observable.subscribe(observer);
4、链式调用
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
emitter.onNext(4);
emitter.onComplete();
}
}).subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Integer integer) {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
5、线程控制
-
subscribeOn：决定发送事件的线程环境，一般传入Schedulers.io()或者Schedulers.newThread(),前者会复用线程池的线程，而后者总会创建一个新线程，所以推荐使用Schedulers.io()； -
observeOn：决定消费事件的线程环境，一般传入AndroidSchedulers.mainThread()，即下游切换回Android的主线程；
完整代码：
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
emitter.onNext(4);
emitter.onComplete();
}
}).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Integer integer) {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
6、总结
-
RxJava是异步请求框架； -
RxJava是扩展的观察者模式，即存在被观察者、观察者、订阅关系； -
RxJava可以灵活切换上下游的线程环境；
