Android :RxJava2.0到底更新了什麼?(含使用建議)
阿新 • • 發佈:2019-01-01
前言
Rxjava
由於其基於事件流的鏈式呼叫、邏輯簡潔 & 使用簡單的特點,深受各大Android
開發者的歡迎。
RxJava 2.0
已於2016 - 10.29正式釋出,對RxJava 1.0
進行了1次重大升級:實際使用的API
及 方法有很大的區別但
RxJava 2.0
的使用思路 和RxJava 1.0
非常類似同時,由於
RxJava 2.0
跟RxJava 1.0
不能共存在1個專案中,所以假如你在使用RxJava 1.0
需要升級到RxJava 2.0
,則需要做一些轉變- 今天,我將為大家帶來
RxJava 2.0
相對於RxJava 1.0
的升級總結 & 從RxJava 1.0
RxJava 2.0
需要注意的坑,希望大家會喜歡
- 本系列文章主要基於
Rxjava 2.0
- 接下來的時間,我將持續推出
Android
中Rxjava 2.0
的一系列文章,包括原理、操作符、應用場景、背壓等等 ,有興趣可以繼續關注Carson_Ho的安卓開發筆記!!
目錄
1. 依賴包更改
- 由於
RxJava 2.0
跟RxJava 1.0
不能共存在1個專案中,所以依賴也不能共存,需要進行更換 - 改動如下
// 原本:`RxJava 1.0` 依賴
compile 'io.reactivex:rxandroid:1.2.0'
compile 'io.reactivex:rxjava:1.1.5'
// 更改:`RxJava 2.0` 依賴
compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
compile 'io.reactivex.rxjava2:rxjava:2.0.7'
// 注:RxJava2 與 RxJava1 不能共存,即依賴不能同時存在
2. 增加被觀察者的新實現:Flowable
- 由於
RxJava 1.0
中 的被觀察者Observable
不能很好地支援背壓(Backpressure
) - 所以,在
RxJava 2.0
中 增加了被觀察者的新實現Flowable
來支援背壓Backpressure
- 而被觀察者的舊實現
Observable
Backpressure
- 而被觀察者的舊實現
3. 建立被觀察者(Observable) & 觀察者(Observer) 方式的區別
在`RxJava 2.0 `中,建立被觀察者(`Observable`) & 觀察者(Observer)的方式也與`RxJava 1.0 `有些區別:- 對於建立被觀察者(
Observable
)
<-- RxJava 1.0 中 建立被觀察者 -->
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();
}
});
<-- RxJava 2.0 中 建立被觀察者 -->
// 變化1:Observable.OnSubscribe介面名改成ObservableOnSubscribe
Observable<Integer> observable=Observable.create(new ObservableOnSubscribe<Integer>() {
// 變化2:複寫的call(Subscriber)改成 subscribe (ObservableEmitter)
// 注:引數也發生了變化,即Subscriber -> ObservableEmitter = 發射器
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
// 可發出三種類型的事件:next事件、complete事件&error事件
// 通過呼叫emitter.onNext(T value) 、onComplete()和onError(Throwable e)
e.onNext(1);
e.onNext(2);
e.onError(new Exception("發生錯誤了"));
e.onComplete();
}
});
- 對於建立 觀察者(
Observer
)
<-- RxJava 1.0 中 建立觀察者(Observer) -->
// 方法1:採用 Observer 介面
Observer<String> observer = new Observer<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!");
}
};
// 方法2:採用 Subscriber 介面(實現了Observer介面的抽象類)
// 與Observer介面的區別:對 Observer介面進行了擴充套件:onStart()、unsubscribe(),但使用方式基本類似
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!");
}
};
<-- RxJava 2.0 中 建立觀察者(Observer) -->
Observer<Integer> observer= new Observer<Integer>() {
// 變化1:增加回調方法onSubscribe()
// 作用:最先呼叫該方法,即適合做初始化工作
@Override
public void onSubscribe(Disposable d) {
// 傳入的引數Disposable作用 類似於 Subsciption
// 即相當於訂閱關係的開關,即可切斷 觀察者和被觀察者的訂閱關係
// 注:呼叫dispose() = 觀察者無法接收事件,但被觀察者還是會繼續傳送事件
}
@Override
public void onNext(Integer value) {
}
@Override
public void onError(Throwable e) {
}
// 變化2:onCompleted()改成 onComplete()
@Override
public void onComplete() {
}
}
4. 簡化訂閱方法
- 對於簡化訂閱的方式,
RxJava 1
主要採用ActionX
介面 &FuncX
介面 - 在
RxJava 2
中,主要是對這一系列介面的名字 按照Java8
的命名規則 進行了修改,而使用方法不變
4.1 ActionX 和 FuncX 改名
- 對於
ActionX
介面名的更改
RxJava 1 | RxJava 2 |
---|---|
Action0 | Action |
Action1 | Consumer(接收1個引數) |
Action2 | BiConsumer (接收2個引數) |
ActionN | Consumer |
<-- 示例1 -->
Disposable disposable = observable.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
//這裡接收資料項
}
}, new Consumer<Throwable>() {
@Override
public void accept(Throwable throwable) throws Exception {
//這裡接收onError
}
}, new Action() {
@Override
public void run() throws Exception {
//這裡接收onComplete。
}
});
<-- 示例2 -->
flowable.subscribe(
new Consumer<String>() {//相當於onNext
@Override
public void accept(String s) throws Exception {
}
}, new Consumer<Throwable>() {//相當於onError
@Override
public void accept(Throwable throwable) throws Exception {
}
}, new Action() {//相當於onComplete,注意這裡是Action
@Override
public void run() throws Exception {
}
}, new Consumer<Subscription>() {//相當於onSubscribe
@Override
public void accept(Subscription subscription) throws Exception {
}
});
4.2 RxJava2的介面方法都允許丟擲異常
即,介面方法里加上了 throws Exception
// Action介面
public interface Action {
void run() throws Exception;
}
// Consumer介面
public interface Consumer<T> {
void accept(T t) throws Exception;
}
// 注:
// 1. 這意味著,在這些方法裡呼叫會發生異常的方法不需要try-catch
// 2. RxJava 2.0 不再支援 null 值,如果傳入一個null會丟擲 NullPointerException
5. 操作符的改變
- 對於操作符,
RxJava 1.0
與RxJava 2.0
在命名 & 行為上大多數保持了一致 - 需要強調的是
first()
、subscribeWith
()和compose()
操作符
5.1 first()操作符
- 改動如下
RxJava 1.0 | RxJava 2.0 |
---|---|
first() | 改名為:firstElement() |
first(Func1) | 棄用,改用為:filter(predicate).first() |
firstOrDefault(T) | 改名為:first(T) |
firstOrDefault(Func1, T) | 改名為:first(T) |
- 示例
<-- RxJava 1.0 -->
Observable
.concat(Observable.from(list))
.first(new Func1<Data, Boolean>() {
@Override
public Boolean call(Data data) {
return DataUtils.isAvailable(data);
}
}).publish();
<-- RxJava 2.0 -->
Observable
.concat(Observable.fromIterable(list))
.filter(new Predicate<Data>() {
@Override
public boolean test(@NonNull Data data) throws Exception {
return DataUtils.isAvailable(data);
}
}).firstElement().toObservable().publish();
5.2 subscribeWith()操作符
具體請看下圖:
5.3 compose()操作符
主要變動在於:
1. RxJava 1.0
實現的是:rx.Observable.Transformer
介面
繼承自
Func1<Observable<T>, Observable<R>>
<-- RxJava 1.0 中的用法 -->
private static <T> Observable.Transformer<T, T> createIOSchedulers() {
return new Observable.Transformer<T, T>() {
@Override
public Observable<T> call(Observable<T> tObservable) {
return tObservable.subscribeOn(Schedulers.io())
.unsubscribeOn(AndroidSchedulers.mainThread())
.observeOn(AndroidSchedulers.mainThread());
}
};
}
public static <T> Observable.Transformer<JsonResult<T>,T> applySchedulers() {
return createIOSchedulers();
}
Action1<Integer> onNext = null;
String[] items = { "item1", "item2", "item3" };
Subscription subscription = Observable.from(items)
.compose(RxUtil.<String>applySchedulers())
.map(new Func1<String, Integer>() {
@Override public Integer call(String s) {
return Integer.valueOf(s);
}
})
.subscribe(onNext);
RxJava 2.0
實現的是io.reactivex.ObservableTansformer<Upstream, Downstream>
一個獨立的介面
<-- RxJava 2.0 中的用法 -->
public static <T> ObservableTransformer<T, T> io2MainObservable() {
return new ObservableTransformer<T, T>() {
@Override
public ObservableSource<T> apply(Observable<T> upstream) {
return upstream.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread());
}
};
}
public static <T> ObservableTransformer<T, T> applySchedulers() {
return io2MainObservable();
}
Consumer<Integer> onNext = null;
String[] items = { "item1", "item2", "item3" };
Disposable disposable = Observable.fromArray(items)
.compose(RxUtil.<String>applySchedulers())
.map(new Function<String, Integer>() {
@Override public Integer apply(String s) throws Exception {
return Integer.valueOf(s);
}
})
.subscribe(onNext);
6. 額外
6.1 新增Processor
作用類似於
Subject
& 繼承自Flowable
= 支援背壓控制而
Subject
則 不支援背壓控制使用如下
//Processor
AsyncProcessor<String> processor = AsyncProcessor.create();
processor.subscribe(o -> Log.d("JG",o)); //three
processor.onNext("one");
processor.onNext("two");
processor.onNext("three");
processor.onComplete();
//Subject
AsyncSubject<String> subject = AsyncSubject.create();
subject.subscribe(o -> Log.d("JG",o));//three
subject.onNext("one");
subject.onNext("two");
subject.onNext("three");
subject.onComplete();
6.2 更改Single
Single
的作用類似於Observable
= 傳送資料,但區別在於訂閱後只能接受到1次- 改動如下
<-- 原始碼分析 -->
// 變動1:Single被重新設計為 Reactive-Streams架構,即SingleSubscriber 改為:SingleObserver
interface SingleObserver<T> {
// 變動2:多了一個回撥方法 onSubscribe()
void onSubscribe(Disposable d);
void onSuccess(T value);
void onError(Throwable error);
}
<-- 具體使用 -->
Single<Long> single = Single.just(1l);
single.subscribe(new SingleObserver<Long>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onSuccess(Long value) {
// 和onNext是一樣的
}
@Override
public void onError(Throwable e) {
}
});
// 注:普通Observable物件可通過toSingle()轉換成Single物件
// 即,Observable.just(1).toSingle()
6.3 更改Completable
Completable
的作用類似於Observable
= 傳送資料,但區別在於訂閱後只能接受Complete
和onError
事件- 改動如下
// 變動1:Completable被重新設計為 Reactive-Streams架構,即CompletableSubscriber 改為:CompletableObserver
interface CompletableObserver<T> {
void onSubscribe(Disposable d);
void onComplete();
void onError(Throwable error);
}
<-- 具體使用 -->
Completable<Long> Completable = Completable.just(1l);
Completable.subscribe(new CompletableObserver<Long>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onComplete(Long value) {
}
@Override
public void onError(Throwable e) {
}
});
// 注:普通Observable物件可通過toCompletable()轉換成Completable物件
// 即,Observable.just(1).toCompletable()
7. 使用建議
對於學習 & 在專案中使用RxJava
的版本選擇,我給出以下建議:
8. 總結
- 本文主要講解了
RxJava 2.0
相對於RxJava 1.0
的變動 從上面可以看到,
RxJava 2.0
相對於RxJava 1.0
最大的改動,主要是增加了被觀察者的新實現:Flowable
& 用於背壓(Backpressure
)的處理接下來的時間,我將持續推出
Android
中Rxjava 2.0
的一系列文章,包括原理、操作符、應用場景、背壓等等 ,有興趣可以繼續關注Carson_Ho的安卓開發筆記!!