Android ReactiveX 新手入門

Getting Started With ReactiveX on Android

前言

在開發 Android 時，常常為了處理網路連線、後端 API 呼叫與 UI 的更新，而寫出一堆 callback，甚至是 nested callbacks，這些程式碼不僅難看，也容易寫錯。使用 ReativeX 可以讓程式碼更清楚明瞭，更可以節省開發的時間。

設定 RxAndroid

compile 'io.reactivex:rxandroid:1.2.1'
// Because RxAndroid releases are few and far between, it is recommended you also
// explicitly depend on RxJava's latest version for bug fixes and new features.
compile 'io.reactivex:rxjava:1.2.1'

PS: 可以到 GitHub 上查看目前最新的版本

基礎：Observables 與 Observers 

當在開發 ReactiveX 時，我們會不斷的使用 observables 和 observers，因為 ReativeX 正正是套用 Observer Design Pattern 的設計，Observable 是可被觀察的對象，也就是真正在做事的人，Observer 是觀察者，聆聽與接收事件發生的人。

建立一個簡單只傳遞出 "Hello" 的 Observable：

Observable<string> myObservable = Observable.just("Hello"); // Emits "Hello"

建立一個 Observer：

Observer<string> myObserver = new Observer<string>() {
    @Override
    public void onCompleted() {
        // Called when the observable has no more data to emit
    }
 
    @Override
    public void onError(Throwable e) {
        // Called when the observable encounters an error
    }
 
    @Override
    public void onNext(String s) {
        // Called each time the observable emits data
        Log.d("OBSERVER1", s);
    }
};

在沒有 Observer 觀察者的時候，Observable 是不會傳遞出資料的，想要 Observable 傳遞出資料，就必須要先向其訂閱 (Subscribe)：

Subscription mySubscription = myObservable.subscribe(myObserver);

此時 myObserver 就會接收到 "Hello" 字串。
除了 observer 可以 subscribe 以外，有時我們不需要 onCompleted 和 onError 時，也可以用 Action1 的介面來 subscribe：

Subscription mySubscription = myObservable
.subscribe(new Action1<string>() {
    
        @Override

        public void call(String s) {
        
            Log.d("ACTION1", s);
    
        }

});

使用 Operators

from

ReactiveX 提供了許多對 observables 操作的運算方法，先來介紹 Observable.from 的用法，from 可以讓你提供一組數據資料，使每個資料都成爲一個 Observable，而 Observer 接受到的資料就會是一個接著一個來，而不是一次收到整組數據。

Observable<Integer> myArrayObservable 
    = Observable.from(new Integer[]{1, 2, 3, 4, 5, 6}); // Emits each item of the array, one at a time
 
myArrayObservable.subscribe(new Action1<Integer>() {
    @Override
    public void call(Integer i) {
        Log.d("My Action", String.valueOf(i)); // Prints the number received
    }
});

map

透過 map 可以將原本 Observable 的結果，做想要的運算，再回傳另一個新的 Observable。

        myArrayObservable
                .map(new Func1<Integer, Integer>() {
                    @Override
                    public Integer call(Integer integer) {
                        return integer * integer;
                    }
                })
                .subscribe(new Action1<Integer>() {
                    @Override
                    public void call(Integer i) {
                        Log.d("ACTION1", i.toString());
                }
        });

skip 

skip 可以用來忽略前面幾個的回傳結果，例如 skip(2) 回傳的結果就會是 3, 4, 5, 6

        myArrayObservable
                .skip(2)
                .subscribe(new Action1<Integer>() {
                    @Override
                    public void call(Integer i) {
                        Log.d("ACTION1", i.toString());
                }
        });

filter 

filter 如同其名，可以用來過濾掉不要的 observables，例如下方的範例就會過濾的奇數的值。

        myArrayObservable
                .filter(new Func1<Integer, Boolean>() {
                    @Override
                    public Boolean call(Integer integer) {
                        if (integer % 2 == 0) {
                            return true;
                        }
                        return false;
                    }
                })
                .subscribe(new Action1<Integer>() {
                    @Override
                    public void call(Integer i) {
                        Log.d("ACTION1", i.toString());
                    }
                });

處理 Asynchronous 任務

剛剛的任務都是執行在 UI Thread(Main Thread) 上的，如果我們要指定 observable 要在哪個 Thread 上執行任務，就要用 subscribeOn，subscribeOn(Schedulers.newThread) 或 subscribeOn(AndroidSchedulers.mainThread)。如果是要指定 observer 在那個 Thread 上接受資料，就要用 observeOn，如果沒有指名 observeOn，observer 就會在 subscribeOn 所指定的 thread 上接收到資料。
只是要傳遞簡單的資料，我們可以用前面介紹的 just 和 from 方法，但大多情況我們都需要執行其他的方法後，才能取得資料，此時就可以用 create 來建立 Observable，並實作 OnSubscribe 介面。

Observable<String> fetchFromGoogle = Observable.create(new Observable.OnSubscribe<String>() {
    @Override
    public void call(Subscriber<? super String> subscriber) {
        try {
            String data = fetchData("http://www.google.com");
            subscriber.onNext(data); // Emit the contents of the URL
            subscriber.onCompleted(); // Nothing more to emit
        }catch(Exception e){
            subscriber.onError(e); // In case there are network errors
        }
    }
});

zip

有時在處理網路任務時，會需要等待兩個以上的任務同時完成，並處理結果，這不但要添加不少程式碼，也容易寫錯，此時 ReactiveX 所提供的 zip 方法便帶來了這樣的好處：

// Fetch from both simultaneously
Observable<String> zipped 
    = Observable.zip(fetchFromGoogle, fetchFromYahoo, new Func2<String, String, String>() {
        @Override
        public String call(String google, String yahoo) { 
            // Do something with the results of both threads
            return google + "\n" + yahoo;
        }
    });

concat

concat 的功能和 from 很像，會把 observable 的結果一個接著一個回傳。

        
Observable observable1 = Observable.just("Hello");
Observable observable2 = Observable.just("World");

// Fetch from both simultaneously
Observable.concat(observable1, observable2)
          .subscribe(new Subscriber<string>() {
              @Override
              public void onCompleted() {}

              @Override
              public void onError(Throwable e) {}

              @Override
              public void onNext(String s) {
                  Log.d("Sean", s);
              }
});

Ref: https://code.tutsplus.com/tutorials/getting-started-with-reactivex-on-android--cms-24387