结构图
RxJava.png
各个类
观察者：
Observer
- onNext() ： 当被订阅时执行
- onCompleted() : 所有onNext() 执行完毕后执行
- onError() : 若onNext()过程出现异常，则执行，与onCompleted()互斥。
Subscriber
implement Observer和Subscription，并添加了onStart()方法。
- onStart() : 在onNext()前调用，可用于数据初始化或做前期准备。总是在subscribe时的线程调用，且不能指定线程，所以不适合进行UI操作。
- unsubscribe() : 解除订阅，不再对Observable的事件进行响应，在适当的适合解除订阅可以避免OOM。
Action0
常用于非显式订阅，因为call中没有参数，常用来充当onCompleted()。
Action1
常用于非显式订阅，因为call中有且仅有一个参数，常用来充当onNext()和onError()。
被观察者
Observable
持有Onsubscribe对象。
可使用observeOn()和subscribeOn()指定执行线程：
- observeOn() 指定Subscriber所运行在的线程。或者叫做事件消费的线程。
- subscribeOn() 指定subscribe()所发生的线程，即 Observable.OnSubscribe被激活时所处的线程。或者叫做事件产生的线程。
参数有：
Schedulers.immediate()/Schedulers.newThread()/Schedulers.io()/Schedulers.computation()/AndroidSchedulers.mainThread()
加载数据常用：
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
Onsubscribe
- call(Subscriber) : 可调用Subscriber中的onNext/onCompleted/onError等方法。
实现方法
- 创建Observable
a. 用Observable.create()，传入OnSubscribe对象，实现里面的call方法。
Observable observable = Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("Hello");
subscriber.onNext("Hi");
subscriber.onNext("Aloha");
subscriber.onCompleted();
}
});
b. 使用操作符just，参数有序，越靠前的参数执行顺序也越靠前。
Observable observable = Observable.just("Hello", "Hi", "Aloha");
c. 使用操作符from，参数为集。
String[] words = {"Hello", "Hi", "Aloha"};
Observable observable = Observable.from(words);
- 创建Observer
a. new Observer OR Subscriber，实现onNext()/onCompleted()/onError()。推荐实现Subscriber。
Subscriber<String> subscriber = new Subscriber<String>() {
@Override
public void onNext(String s) {
Log.d(tag, "Item: " + s);
}
@Override
public void onCompleted() {
Log.d(tag, "Completed!");
}
@Override
public void onError(Throwable e) {
Log.d(tag, "Error!");
}
};
b. 不完整回调，可只创建其中某个或某些。
Action1<String> onNextAction = new Action1<String>() {
// onNext()
@Override
public void call(String s) {
Log.d(tag, s);
}
};
Action1<Throwable> onErrorAction = new Action1<Throwable>() {
// onError()
@Override
public void call(Throwable throwable) {
// Error handling
}
};
Action0 onCompletedAction = new Action0() {
// onCompleted()
@Override
public void call() {
Log.d(tag, "completed");
}
};
- 订阅
a. new Observer OR Subscriber，则：
observable.subscribe(observer);
// 或者：
observable.subscribe(subscriber);
b. 不完整定义，会自动创建Subscribe，并识别参数定义相关方法。
observable.subscribe(onNextAction);
observable.subscribe(onNextAction, onErrorAction);
observable.subscribe(onNextAction, onErrorAction, onCompletedAction);
