Rxjava

Rxjava配置

compile 'io.reactivex:rxandroid:1.2.1' compile 'io.reactivex:rxjava:1.1.6' 使用基本流程Observable创建 Observer创建

订阅

Observable的创建

create操作符(创建observable的基本方法)

// 参数为OnSubscribe类型 public static <T> Observable<T> create(OnSubscribe<T> f) { return new Observable<T>(hook.onCreate(f)); } OnSubscribe是Observable的内部类并且继承于Action接口 public interface OnSubscribe<T> extends Action1<Subscriber<? super T>> { // cover for generics insanity }

具体用法

Observable<String> observable = Observable.create(new Observable.OnSubscribe<String>() { @Override public void call(Subscriber<? super String> subscriber) { //调用subsc()方法传入的subscriber subscriber.onNext("test"); subscriber.onCompleted(); } });

订阅

订阅Observer

observable.subscribe(new Observer<String>() { @Override public void onCompleted() { } @Override public void onError(Throwable e) { } @Override public void onNext(String s) { Log.i("RxJavaTest", "RxJavaTest s = " + s); } });

订阅Subscriber

observable.subscribe(new Subscriber<String>() { @Override public void onCompleted() { } @Override public void onError(Throwable e) { } @Override public void onNext(String s) { Log.i("RxJavaTest", "RxJavaTest s = " + s); } });

其它订阅写法

observable.subscribe(new Action1<String>() { @Override public void call(String s) { Log.i("RxJavaTest", "RxJavaTest s = " + s); } });

三种订阅方法区别

三种订阅方法的本质都是订阅Subscriber，订阅Observer会主动创建一个叫ObserverSubscriber的对象，订阅Action1回创建一个叫ActionSubscriber的对象

public final Subscription subscribe(final Observer<? super T> observer) { //判断Observer是否为Subscriber， 是：直接订阅，否：创建新的Subscriber再订阅 if (observer instanceof Subscriber) { return subscribe((Subscriber<? super T>)observer); } return subscribe(new ObserverSubscriber<T>(observer)); } public final Subscription subscribe(final Action1<? super T> onNext) { if (onNext == null) { throw new IllegalArgumentException("onNext can not be null"); } Action1<Throwable> onError = InternalObservableUtils.ERROR_NOT_IMPLEMENTED; Action0 onCompleted = Actions.empty(); //创建一个新的Subscriber并把Action1当作参数传入 return subscribe(new ActionSubscriber<T>(onNext, onError, onCompleted)); }

Observer与Subscriber的区别

Observer是接口，它的直接子类是SubscriberSubscriber及实现了Observer，又实现接口Subscription，Subscription中的unsubscribe()方法和isUnsubscribed()分别是取消订阅和判断订阅是否已取消 public abstract class Subscriber<T> implements Observer<T>, Subscription {}

subscribe订阅的基本流程(精简方法，只有流程)

调用onSubscribe的call方法，传入订阅着subscriber static <T> Subscription subscribe(Subscriber<? super T> subscriber, Observable<T> observable) { //调用订阅者的onStart()方法 subscriber.onStart(); //调用OnSubscribe接口的call方法，onSubscribe为创建observable时传入的OnSubscribe对象 onSubscribe.call(subscriber); }

在onSubscribe的call方法中具体分发事件，也就是subscriber的onNext,onComplete和onError方法

Observable.OnSubscribe<String> onSubscribe = new Observable.OnSubscribe<String>() { @Override public void call(Subscriber<? super String> subscriber) { //subscriber为调用subscribe方法是传入的subscriber，也就是订阅该事件的订阅者，subscriber为父类，具体的类型根据订阅时传入的参数不同而不同，例如订阅时传入的Action接口类型，此时的subscriber为ActionSubscriber类型 subscriber.onNext("test"); subscriber.onCompleted(); } }; //创建observable，传入onSubscribe Observable<String> observable = Observable.create(onSubscribe);

在订阅方法中出入的参数为Action接口时，会创建ActionSubscriber对象，并把Action做为参数传入新创建的对象，新创建ActionSubscriber对象会重写Subscriber的OnNext方法

//重写Subscriber的OnNext方法 @Override public void onNext(T t) { //onNext为调用subscribe时传入的Action接口对象 onNext.call(t); }