前言

剛參加工作的時候接觸到了專案中的Rxjava，當時一點基礎沒有，學習了好長時間才漸漸學會使用，但也只是皮毛，停留在表面。後來換工作了發現Rxjava對找工作的幫助還是挺大的，因為是流行框架，都可以講給面試官聽，體現出自己追求技術的精神。但實際上，學會Rxjava對技術的提升還是很有幫助的，所以我會為大家講解Rxjava的相關知識，以及Rxjava 1 到Rxjava 2的變化，幫助大家能夠學會使用Rxjava，並能在面試中征服面試官！
本文的主要作用是：如果之前有點基礎，那就是複習鞏固，順便了解一下新版本的Rxjava的變化；如果之前沒有基礎，那就來認識一下Rxjava,

Rxjava是什麼？

RxJava 在官網上的說明是：

a library for composing asynchronous and event-based programs using observable sequences for the Java VM
一個在 Java VM上使用可觀測的序列來組成非同步的、基於事件的程式的庫

這句話很難懂，但是可以概括成一個核心詞——非同步

Rxjava主要作用就是用來處理非同步，當你的業務需要訪問資料庫，訪問網路，或者任何耗時的操作，都可以藉助Rxjava來實現。
但是有人說在Android中已經有很多非同步操作的API，比如Handler，AsyncTask等，這些都能滿足基本的非同步操作，為什麼還要使用Rxjava呢？
首先我們開看一個例子做個比較：

假設有這樣一個需求：介面上有一個自定義的檢視 imageCollectorView ，它的作用是顯示多張圖片，並能使用 addImage(Bitmap) 方法來任意增加顯示的圖片。現在需要程式將一個給出的目錄陣列 File[] folders 中每個目錄下的 png 圖片都加載出來並顯示在 imageCollectorView中。需要注意的是，由於讀取圖片的這一過程較為耗時，需要放在後臺執行，而圖片的顯示則必須在 UI 執行緒執行。

常用的實現方式有多種，比如：
程式碼塊1

//採用android自帶的api實現
new Thread() {
    @Override
    public
 
 void run() {
        super.run();
        for (File file : files) {
            File[] files = folder.listFiles();
            for (File file : files) {
                if (file.getName().endsWith(".png")) {
                    final Bitmap bitmap = getBitmapFromFile(file);
                    getActivity().runOnUiThread(new Runnable() {
                        @Override
                        public void run() {
                            imageCollectorView.addImage(bitmap);
                        }
                    });
                }
            }
        }
    }
}.start();

如果使用Rxjava，則可以這樣實現：
程式碼塊2

//採用Rxjava實現
Observable.from(folders)
    .flatMap(new Func1<File, Observable<File>>() {
        @Override
        public Observable<File> call(File file) {
            return Observable.from(file.listFiles());
        }
    })
    .filter(new Func1<File, Boolean>() {
        @Override
        public Boolean call(File file) {
            return file.getName().endsWith(".png");
        }
    })
    .map(new Func1<File, Bitmap>() {
        @Override
        public Bitmap call(File file) {
            return getBitmapFromFile(file);
        }
    })
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new Action1<Bitmap>() {
        @Override
        public void call(Bitmap bitmap) {
            imageCollectorView.addImage(bitmap);
        }
    });

雖然程式碼塊2的程式碼量比程式碼塊1的程式碼量要多，但是很明顯程式碼塊2的程式碼看起來更整潔更優雅，而且如果讀者學過Rxjava的人，會明顯感覺到程式碼塊2的可讀性比程式碼塊1的可讀性要強。
由此可見Rxjava的優點即是：採用鏈式呼叫，程式碼簡潔優雅有美感，並且可讀性增強！

以上，是Rxjava的一部分優點，其實Rxjava的優點更在於它的強大。

下面我們簡單瞭解一下Rxjava的原理：

Rxjava實現非同步的方法是通過觀察者模式實現的。

什麼事觀察者模式呢？

舉個例子，使用者介面可以作為一個觀察者，業務資料是被觀察者，使用者介面觀察業務資料的變化，發現數據變化後，就顯示在介面上。

在Android中最常見的觀察者模式是View的onClick事件模型。

如圖可見，當Button持有OnClickListener物件之後，Button被點選之後會自動觸發OnClickListener中的OnClick方法。
把上面的Button點選事件抽象一下就變成：

當Observable(可觀察的，被觀察者)的狀態發生變化時，內部會通過一系列事件觸發Observer(觀察者)中的方法，可以做出相應的操作。
可能這樣講還是比較抽象，舉個簡單的生活中的例子：

以上模型中，上課鈴聲是被觀察者，即Observable，可觀察的，被觀察者；學生就是觀察者，即Observer（觀察者）,學生聽到上課鈴聲響了，就會去上課，這就是學生根據上課鈴聲所做出的反應。
也就是：
被觀察者狀態發生變化，觀察者可以做出反應。

在Rxjava中觀察者模式

RxJava 有四個基本概念：Observable (可觀察者，即被觀察者)、 Observer (觀察者)、 subscribe (訂閱)、事件。Observable 和 Observer 通過 subscribe() 方法實現訂閱關係，從而 Observable 可以根據情況回撥來通知 Observer。
Rxjava常用的的回撥方法有三種：
- onNext：完成佇列中的一個事件
- onComplete：完成佇列中所有的事件
- onError：事件發生錯誤時，並且後續的事件終止。

為什麼Rxjava要使用觀察者模式呢？
因為觀察者模式在模組之間劃定了清晰的界限，降低模組耦合性，提高了程式碼的可維護性和重用性。

Rxjava基本使用方法

1. 建立Observer
   Observer是觀察者，當被觀察者狀態發生變化的時候，他會收到相應的事件，使用者可以根據不同的事件進行不同的處理。

Observer<String> observer = new Observer<String>() {
            @Override
            public void onCompleted() {
                Log.d("Rxjava demo", "onCompleted");
            }

            @Override
            public void onError(Throwable e) {
                Log.d("Rxjava demo", "onError");
            }

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

其實，除了使用Observer以外，Rxjava還有個Subscriber。這個是實現了Observer的抽象類，裡面對Observer進行了一些擴充套件。

 Subscriber<String> subscriber = new Subscriber<String>() {

            @Override
            public void onStart() {
                super.onStart();
            }

            @Override
            public void onNext(String s) {
                Log.d("Rxjava demo", "Item: " + s);
            }

            @Override
            public void onCompleted() {
                Log.d("Rxjava demo", "Completed!");
            }

            @Override
            public void onError(Throwable e) {
                Log.d("Rxjava demo", "Error!");
            }
        };

可以看出，Subscriber比Observer多了一個回撥方法onStart()，它會在事件開始執行之前的時候呼叫，用於做一些準備工作，類似於AsyncTask中的onPreExecute方法。
但是subscriber中還有幾個很重要的方法：
- unsubscribe()：這個方法是取消訂閱事件，一般有利於防止記憶體洩漏。在android開發中我們知道一般有訂閱就應該有取消訂閱。
- isUnsubscribed()：這個方法是用於判斷事件是否被訂閱。
- add(Subscription s)：這個方法是把一個Subscription 新增到Subscription列表中，便於統一管理，取消訂閱等
2. 建立Observable

Observable observable = Observable.create(new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> subscriber) {
                subscriber.onNext("onNext");
                subscriber.onCompleted();
                subscriber.onNext("onNext");
                subscriber.onError(new Throwable());
            }
        });

使用create方法建立Observable（被觀察者），然後call方法會被自動呼叫，在call方法內部定義事件的回撥的行為。
其實這段程式碼中，當執行了onComplete方法之後，就不會在往下執行了，也就是說onError方法不會被呼叫，因為事件已經完全執行完成，就會停止執行之後的事件。
如果我們反過來寫：

Observable observable = Observable.create(new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> subscriber) {
                subscriber.onNext("onNext");
                subscriber.onError(new Throwable());
                subscriber.onCompleted();
                subscriber.onNext("onNext");
            }
        });

onNext方法執行完成之後會執行onError，但是之後的onComplete方法以及後面的事件都不會在執行了，前面我們說過，onError執行之後表示事件執行失敗，後面的事件就會停止執行。
3. Subscribe(訂閱)

observable.subscribe(observer);

最後我們使用subscribe方法讓observer訂閱observable。但是這個方法看起來寫反了，他不是“觀察者”訂閱“被觀察者”，而是被觀察者訂閱了觀察者，這其實是因為為了保證流式的設計，把subscribe是Observable的方法，把observer作為引數傳進。
什麼保證流式設計呢？
因為Rxjava可以這樣寫：

Observable.create(new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> subscriber) {
                subscriber.onNext("onNext");
                subscriber.onCompleted();
                subscriber.onError(new Throwable());
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onCompleted() {
                Log.d("Rxjava demo", "onCompleted");
            }

            @Override
            public void onError(Throwable e) {
                Log.d("Rxjava demo", "onError");
            }

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

這樣程式碼就會看著優雅許多，而且層級清晰，可讀性強。
通過以上方法，我們就簡單瞭解了Rxjava的使用方法。但其實，Observable的建立方式有多種:
- 例如just可以傳入多個引數，最多可以傳入10個引數，並且會自動呼叫10次onNext
- from(T[])將傳入的陣列依次傳送出去，陣列內有多少個元素，就會呼叫多少次onNext，當所有元素（事件）傳送結束之後會呼叫onComplete，如果在某個元素中發生錯誤，就會呼叫onError。
寫法如下：

ArrayList<String> array = new ArrayList<>();
Observable.from(array).subscribe(new Subscriber<String>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(String s) {

            }
        });

大家可以看到，不用再重寫call方法，因為會自動安排事件傳送，不需要手動呼叫onNext等方法了，而這段程式碼中的onNext方法會依次輸出陣列中的每一個元素。

Rxjava的變化

以上內容都是基於Rxjava比較舊的API介紹的，目前Rxjava 1 已經更新到了1.3
使用最新的Rxjava 1需要引入以下依賴：

compile 'io.reactivex:rxjava:1.3.0'
compile 'io.reactivex:rxandroid:1.2.1'

在這個版本中的Observable的建立有所變化,方法 static Observable create(OnSubscribe f)已經過時了，因為這個方法不安全。
新的版本中已經引入了比較安全的方法：
- static

 Observable.create(new SyncOnSubscribe<String, String>() {
            @Override
            protected String generateState() {
                Log.d("Rxjava demo", "generateState");
                return "generateState";
            }

            @Override
            protected String next(String state, Observer<? super String> observer) {
                observer.onNext(state);
                observer.onCompleted();
                observer.onError(new Throwable("onError"));
                return state;
            }
        }).subscribe(new Action1<String>() {
            @Override
            public void call(String s) {
                Log.d("Rxjava demo", s);
            }
        }, new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                Log.d("Rxjava demo", throwable.getMessage());
            }
        }, new Action0() {
            @Override
            public void call() {
                Log.d("Rxjava demo", "onComplete");
            }
        });

但是可以看出來多了兩種回撥方法：
- generateState()，這個方法會在subscribe的時候呼叫，產生一個state值，這個值會在第一次迭代的時候傳遞到next(S state, Observer observer) 方法中，後續迭代下將收到由先前的呼叫返回下一個狀態。也就是會收到next(S state, Observer observer)的返回值

• next(S state, Observer observer)中會收到上游傳來的資料，並通過observer.onNext方法傳遞到下游。但是該方法的實現必須遵循以下規則：(1)observer.onNext(t)不能超過1次呼叫。(2)不能同時呼叫observer.onNext(t)。
  next(S state, Observer observer)會返回下一次迭代的狀態值（state）給generateState()，然後generateState()再把值傳遞給next(S state, Observer observer)，如果你沒有呼叫onComplete或者onError，這個迴圈會一直下去

好了，Rxjava簡單的介紹就到這裡了，下次我們會介紹Rxjava最強大的地方，也就是Rxjava操作符。