Android RxJava1.X升级到RxJava2.X笔记

简书地址 http://www.jianshu.com/p/2badfbb3a33b

描述	RxJava 1.X	RxJava 2.X
`package`包名	`rx.xxx`	`io.reactivex.xxx`
Reactive Streams规范	`1.X`早于`Reactive Streams`规范出现，仅部分支持规范	完全支持
Backpressure 背压	对背压的支持不完善	`Observable`设计为不支持背压新增`Flowable`支持背压
`null`空值	支持	不再支持`null`值，传入`null`值会抛出 `NullPointerException`
`Schedulers`线程调度器	`Schedulers.immediate()` `Schedulers.trampoline()` `Schedulers.computation()` `Schedulers.newThread()` `Schedulers.io()` `Schedulers.from(executor)` `AndroidSchedulers.mainThread()`	移除`Schedulers.immediate()` 新增`Schedulers.single()` 其它未变
`Single`	行为类似`Observable`，但只会发射一个`onSuccess`或`onError`	按照`Reactive Streams`规范重新设计，遵循协议`onSubscribe(onSuccess/onError)`
`Completable`	行为类似`Observable`，要么全部成功，要么就失败	按照`Reactive Streams`规范重新设计，遵循协议`onSubscribe (onComplete/onError)`
`Maybe`	无	`2.X`新增，行为类似`Observable`，可能会有一个数据或一个错误，也可能什么都没有。可以将其视为一种返回可空值的方法。这种方法如果不抛出异常的话，将总是会返回一些东西，但是返回值可能为空，也可能不为空。按照`Reactive Streams`规范设计，遵循协议`onSubscribe (onSuccess/onError/onComplete)`
`Flowable`	无	`2.X`新增，行为类似`Observable`，按照`Reactive Streams`规范设计，支持背压`Backpressure`
`Subject`	`AsyncSubject` `BehaviorSubject` `PublishSubject` `ReplaySubject` `UnicastSubject`	`2.X`依然维护这些`Subject`现有的功能，并新增： `AsyncProcessor` `BehaviorProcessor` `PublishProcessor` `ReplayProcessor` `UnicastProcessor` 支持背压`Backpressure`
`Subscriber`	`Subscriber`	由于与`Reactive Streams`的命名冲突，`Subscriber`已重命名为`Disposable`

RxJava 2.X + Retrofit + OkHttp 简单示例点这里

`library`依赖变化

//1.X
compile 'io.reactivex:rxjava:1.2.1'
compile 'io.reactivex:rxandroid:1.2.1'

//2.X
compile 'io.reactivex.rxjava2:rxjava:2.0.0'
compile 'io.reactivex.rxjava2:rxandroid:2.0.0'

`package`变化

变动主要为rx.xxx --> io.reactivex.xxx

//1.X
import rx.Observable;
import rx.Subscription;
import rx.android.schedulers.AndroidSchedulers;
import rx.schedulers.Schedulers;
import rx.functions.Action1;

//2.X
import io.reactivex.Observable;
import io.reactivex.ObservableSource;
import io.reactivex.ObservableTransformer;
import io.reactivex.disposables.Disposable;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.schedulers.Schedulers;
import io.reactivex.functions.Consumer;

`null`值

RxJava 2.X不再支持null值，如果传入一个null会抛出NullPointerException

Observable.just(null);

Single.just(null);

Observable.fromCallable(() -> null)
    .subscribe(System.out::println, Throwable::printStackTrace);

Observable.just(1).map(v -> null)
    .subscribe(System.out::println, Throwable::printStackTrace);

案例1

//1.X
public static final Observable.Transformer IO_TRANSFORMER = new Observable.Transformer() {
    @Override public Object call(Object observable) {
        return ((Observable) observable).subscribeOn(Schedulers.io())
                                        .unsubscribeOn(Schedulers.io())
                                        .observeOn(Schedulers.io());
    }
};
public static final <T> Observable.Transformer<T, T> applySchedulers(Observable.Transformer transformer){
    return (Observable.Transformer<T, T>)transformer;
}
Action1<Integer> onNext = null;
String[] items = { "item1", "item2", "item3" };
Subscription subscription = Observable.from(items)
                                      .compose(RxUtil.<String>applySchedulers(IO_TRANSFORMER))
                                      .map(new Func1<String, Integer>() {
                                                  @Override public Integer call(String s) {
                                                      return Integer.valueOf(s);
                                                  }
                                              })
                                      .subscribe(onNext);
//TODO subscription.unsubscribe();   

//2.X
public static final ObservableTransformer IO_TRANSFORMER = new ObservableTransformer() {
    @Override public ObservableSource apply(Observable upstream) {
        return upstream.subscribeOn(Schedulers.io())
                       .unsubscribeOn(Schedulers.io())
                       .observeOn(Schedulers.io());
    }
};
public static final <T> ObservableTransformer<T, T> applySchedulers(ObservableTransformer transformer){
    return (ObservableTransformer<T, T>)transformer;
}
Consumer<Integer> onNext = null;
String[] items = { "item1", "item2", "item3" };
Disposable disposable = Observable.fromArray(items)
                                  .compose(RxUtil.<String>applySchedulers(IO_TRANSFORMER))
                                  .map(new Function<String, Integer>() {
                                              @Override public Integer apply(String s) throws Exception {
                                                  return Integer.valueOf(s);
                                              }
                                          })
                                  .subscribe(onNext);
//TODO disposable.dispose();

.subscribe(...)返回值的变化:1.X为Subscription, 2.X为Disposable
Transformer的变化:1.X为rx.Observable内部的Transformer接口, 继承自Func1<Observable<T>, Observable<R>>, 2.X为io.reactivexObservableTransformer<Upstream, Downstream>,是一个独立的接口
AndroidSchedulers的变化: 1.X为rx.android.schedulers.AndroidSchedulers, 2.X为io.reactivex.android.schedulers.AndroidSchedulers
Func1的变化: 1.X为rx.functions.Func1, 2.X为io.reactivex.functions.Function
其它重载方法见下方截图

//1.X
public class AppBaseActivity extends AppCompatActivity {
    ...
    private CompositeSubscription mCompositeSubscription;

    protected void addSubscription(Subscription subscription) {
        if (null == mCompositeSubscription) {
            mCompositeSubscription = new CompositeSubscription();
        }
        mCompositeSubscription.add(subscription);
    }

    @Override protected void onDestroy() {
        if (null != mCompositeSubscription) {
            mCompositeSubscription.unsubscribe();
        }
        super.onDestroy();
    }
    ...
}

//2.X
public class AppBaseActivity extends AppCompatActivity {
    ...
    private   CompositeDisposable mCompositeDisposable;

    protected void addDisposable(Disposable disposable) {
        if (null == mCompositeDisposable) {
            mCompositeDisposable = new CompositeDisposable();
        }
        mCompositeDisposable.add(disposable);
    }

    @Override protected void onDestroy() {
        if (null != mCompositeDisposable) {
            mCompositeDisposable.clear();
        }
        super.onDestroy();
    }
    ...
}

library依赖变化

package变化

null值

`library`依赖变化

`package`变化

`null`值