基本結構
我們先來看一段最基本的代碼，分析這段代碼在RxJava中是如何實現的。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
|
Observable.OnSubscribe<String> onSubscriber1 = new Observable.OnSubscribe<String>() { @Override public void call(Subscriber<? super String> subscriber) { subscriber.onNext("1"); subscriber.onCompleted(); }};Subscriber<String> subscriber1 = new Subscriber<String>() { @Override public void onCompleted() { } @Override public void onError(Throwable e) { } @Override public void onNext(String s) { }};Observable.create(onSubscriber1) .subscribe(subscriber1); |
首先我們來看一下Observable.create的代碼
|
1
2
3
4
5
6
7
|
public final static <T> Observable<T> create(OnSubscribe<T> f) { return new Observable<T>(hook.onCreate(f));}protected Observable(OnSubscribe<T> f) { this.onSubscribe = f;} |
直接就是調用了Observable的構造函數來創建一個新的Observable對象，這個對象我們暫時標記為observable1，以便后面追溯。
同時，會將我們傳入的OnSubscribe對象onSubscribe1保存在observable1的onSubscribe屬性中，這個屬性在后面的上下文中很重要，大家留心一下。
接下來我們來看看subscribe方法。
|
1
2
3
4
5
6
7
8
9
10
|
public final Subscription subscribe(Subscriber<? super T> subscriber) { return Observable.subscribe(subscriber, this);}private static <T> Subscription subscribe(Subscriber<? super T> subscriber, Observable<T> observable) { ... subscriber.onStart(); hook.onSubscribeStart(observable, observable.onSubscribe).call(subscriber); return hook.onSubscribeReturn(subscriber);} |
可以看到，subscribe之后，就直接調用了observable1.onSubscribe.call方法，也就是我們代碼中的onSubscribe1對象的call方法
，傳入的參數就是我們代碼中定義的subscriber1對象。call方法中所做的事情就是調用傳入的subscriber1對象的onNext和onComplete方法。
這樣就實現了觀察者和被觀察者之間的通訊，是不是很簡單？
|
1
2
3
4
|
public void call(Subscriber<? super String> subscriber) { subscriber.onNext("1"); subscriber.onCompleted();} |
RxJava使用場景小結
1.取數據先檢查緩存的場景
取數據，首先檢查內存是否有緩存
然后檢查文件緩存中是否有
最后才從網絡中取
前面任何一個條件滿足，就不會執行后面的
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
|
final Observable<String> memory = Observable.create(new Observable.OnSubscribe<String>() { @Override public void call(Subscriber<? super String> subscriber) { if (memoryCache != null) { subscriber.onNext(memoryCache); } else { subscriber.onCompleted(); } }});Observable<String> disk = Observable.create(new Observable.OnSubscribe<String>() { @Override public void call(Subscriber<? super String> subscriber) { String cachePref = rxPreferences.getString("cache").get(); if (!TextUtils.isEmpty(cachePref)) { subscriber.onNext(cachePref); } else { subscriber.onCompleted(); } }});Observable<String> network = Observable.just("network");//主要就是靠concat operator來實現Observable.concat(memory, disk, network).first().subscribeOn(Schedulers.newThread()).subscribe(s -> { memoryCache = "memory"; System.out.println("--------------subscribe: " + s);}); |
2.界面需要等到多個接口并發取完數據，再更新
|
1
2
3
4
5
6
7
8
9
|
//拼接兩個Observable的輸出，不保證順序，按照事件產生的順序發送給訂閱者private void testMerge() { Observable<String> observable1 = DemoUtils.createObservable1().subscribeOn(Schedulers.newThread()); Observable<String> observable2 = DemoUtils.createObservable2().subscribeOn(Schedulers.newThread()); Observable.merge(observable1, observable2) .subscribeOn(Schedulers.newThread()) .subscribe(System.out::println);} |
3.一個接口的請求依賴另一個API請求返回的數據
舉個例子，我們經常在需要登陸之后，根據拿到的token去獲取消息列表。
這里用RxJava主要解決嵌套回調的問題，有一個專有名詞叫Callback hell
|
1
2
3
4
5
|
NetworkService.getToken("username", "password") .flatMap(s -> NetworkService.getMessage(s)) .subscribe(s -> { System.out.println("message: " + s); }); |
4.界面按鈕需要防止連續點擊的情況
|
1
2
3
4
5
|
RxView.clicks(findViewById(R.id.btn_throttle)) .throttleFirst(1, TimeUnit.SECONDS) .subscribe(aVoid -> { System.out.println("click"); }); |
5.響應式的界面
比如勾選了某個checkbox，自動更新對應的preference
|
1
2
3
4
5
6
7
8
|
SharedPreferences preferences = PreferenceManager.getDefaultSharedPreferences(this);RxSharedPreferences rxPreferences = RxSharedPreferences.create(preferences);Preference<Boolean> checked = rxPreferences.getBoolean("checked", true);CheckBox checkBox = (CheckBox) findViewById(R.id.cb_test);RxCompoundButton.checkedChanges(checkBox) .subscribe(checked.asAction()); |
6.復雜的數據變換
|
1
2
3
4
5
6
7
|
Observable.just("1", "2", "2", "3", "4", "5") .map(Integer::parseInt) .filter(s -> s > 1) .distinct() .take(3) .reduce((integer, integer2) -> integer.intValue() + integer2.intValue()) .subscribe(System.out::println);//9 |
