什么是RxJava?

要知道什么是RxJava首先要去了解下Rx。Rx全稱Reactive Extensions,直譯過來就是響應式擴展。
RxJava是基于觀察者模式，他是一種編程模型，目標是提供一致的編程接口，幫助開發(fā)者方便的處理異步數(shù)據(jù)流。

ReactiveX.io給的定義是，Rx是一個使用可觀察數(shù)據(jù)流進行異步編程的編程接口，ReactiveX結合了觀察者模式、迭代器模式和函數(shù)式編程的精華。Rx已經(jīng)滲透到了各個語言中，有了Rx所以才有了 RxJava，Rx.NET、RxJS、RxSwift、Rx.rb、RxPHP等等，更詳細的可以去這里看看languages

那么RxJava到底是什么呢?我對他的理解就是對于java語言的異步的響應式編程庫。

簡單說下什么是響應式編程

舉個簡單的栗子；

    A=B+C

A被賦值為B和C的和，這是如果B的值發(fā)生變化，A的值并不會發(fā)生變化。而如果我們運用一種機制，當B或者C發(fā)送變化，A的值也隨之變化，這就是響應式。

關于響應式的更多更多了解點擊這里傳送門

聊聊RxJava的觀察者模式

剛剛已經(jīng)了解了RxJava是有RX衍生過來的，所以RxJava也是基于觀察者模式。就好比一個害羞的小男孩，暗戀一個小女孩，女生的一舉一動他都認真仔細的觀察，女孩餓了，渴了。。。最后終于看著女孩被別人牽走啦。所以不能只說觀察，也要行動啊。

上例中的女孩就是被觀察者，男孩則是觀察者。對應RxJava中，有這幾個基本概念:

• Observable (可觀察者，即被觀察者)
• Observer (觀察者)
• subscribe (訂閱)
• 事件

Observable 和 Observer 通過 subscribe() 方法實現(xiàn)訂閱關系，從而 Observable 可以在需要的時候發(fā)出事件來通知 Observer。

與傳統(tǒng)觀察者模式不同， RxJava 的事件回調方法除了普通事件 onNext() 
（相當于 onClick() / onEvent()）之外，還定義了兩個特殊的事件：onCompleted() 和 onError()。

onCompleted(): 事件隊列完結。RxJava 不僅把每個事件單獨處理，還會把它們看做一個隊列。
RxJava 規(guī)定，當不會再有新的 onNext() 發(fā)出時，需要觸發(fā) onCompleted() 方法作為標志。

onError(): 事件隊列異常。在事件處理過程中出異常時，onError() 會被觸發(fā)，同時隊列自動終止，不允許再有事件發(fā)出。

在一個正確運行的事件序列中, onCompleted() 和 onError() 有且只有一個，并且是事件序列中的最后一個。需要注意的是，onCompleted() 和 onError() 二者也是互斥的，即在隊列中調用了其中一個，就不應該再調用另一個。

RxJava API的使用

(1) 創(chuàng)建 Observer

observer_create.png

(2) 創(chuàng)建 Observable

observable_create.png

也可以利用just或者from快速創(chuàng)建observable

• just(T...): 將傳入的參數(shù)依次發(fā)送出來。
  Observable observable = Observable.just("Hello", "Hi", "Aloha");
  // 將會依次調用：
  // onNext("Hello");
  // onNext("Hi");
  // onNext("Aloha");
  // onCompleted();
• from(T[]) / from(Iterable<? extends T>) : 將傳入的數(shù)組或 Iterable 拆分成具體對象后，依次發(fā)送出來。
  String[] words = {"Hello", "Hi", "Aloha"};
  Observable observable = Observable.from(words);
  // 將會依次調用：
  // onNext("Hello");
  // onNext("Hi");
  // onNext("Aloha");
  // onCompleted();

(3) Subscribe (訂閱)

創(chuàng)建了 Observable 和 Observer 之后，再用 subscribe() 方法將它們聯(lián)結起來，整條鏈子就可以工作了。代碼形式很簡單：

observable.subscribe(observer);
// 或者：
observable.subscribe(subscriber);

(4)線程控制Scheduler

Schedulers.png

在調用Scheduler的時候有這兩個方法

• subscribeOn() : 設置的是被觀察者所做的一些操作的線程(發(fā)送事件的線程)
• observeOn() : 設置的是觀察者被通知之后在哪個線程操作(接收事件的線程)

多次指定發(fā)送事件(上游)的線程只有第一次指定的有效, 也就是說多次調用subscribeOn() 只有第一次的有效, 其余的會被忽略.

多次指定接收(下游)的線程是可以的, 也就是說每調用一次observeOn() , 下游的線程就會切換一次.

RxJava 操作符

• map
  Map操作符對原始Observable發(fā)射的每一項數(shù)據(jù)應用一個你選擇的函數(shù)，然后返回一個發(fā)射這些結果的Observable。

map.png

例：得到多個Student對象中的name，保存到nameList中

    Observable.just(student1, student2, student2)
                    //使用map進行轉換，參數(shù)1：轉換前的類型，參數(shù)2：轉換后的類型
                    .map(new Func1<Student, String>() {
                        @Override
                        public String call(Student i) {
                            String name = i.getName();//獲取Student對象中的name
                            return name;//返回name
                        }
                    })
                    .subscribe(new Action1<String>() {
                        @Override
                        public void call(String s) {
                            nameList.add(s);
                        }
                    });

上例，map將原來的student數(shù)據(jù)源轉化為String name 發(fā)送出去了，觀察者拿到的就是name

• FlatMap
  FlatMap將一個發(fā)射數(shù)據(jù)的Observable變換為多個Observables，然后將它們發(fā)射的數(shù)據(jù)合并后放進一個單獨的Observable

• concatMap
  它類似于最簡單版本的flatMap，但是它按次序連接而不是合并那些生成的Observables，然后產(chǎn)生自己的數(shù)據(jù)序列。

flatMap.png

例如，上面是打印每個學生的名字,這邊我們需要打印出每個學生的課程名稱，一個學生有多個課程

      /**
         * 學生類
         */
        class Student {
            private String name;//姓名
            private List<Course> coursesList;//所修的課程
            ...
        }
        /**
         * 課程類
         */
        class  Course {
            private String name;//課程名
            private String id;
            ...
        }

利用flatMap變化一下:

     List<Student> students = new ArrayList<Student>();
            students.add...
            ...

            Observable.from(students)
                    .flatMap(new Func1<Student, Observable<Course>>() {
                        @Override
                        public Observable<Course> call(Student student) {
                            return Observable.from(student.getCoursesList());
                        }
                    })
                    .subscribe(new Action1<Course>() {
                        @Override
                        public void call(Course course) {
                            Log.i(TAG, course.getName());
                        }
                    });

上述例子中，數(shù)據(jù)源轉化是這樣的:

 student  ->  course1,course2...  -> 利用from()將所有的course放入一個observerble中發(fā)送出去

 接收消息的數(shù)據(jù)源就變成 List<Course>課程的列表,直接打印名稱

• zip
  通過一個函數(shù)將多個Observables的發(fā)射物結合到一起，基于這個函數(shù)的結果為每個結合體發(fā)射單個數(shù)據(jù)項。

zip.png

使用場景:
比如一個界面需要展示用戶的一些信息, 而這些信息分別要從兩個服務器接口中獲取, 而只有當兩個都獲取到了之后才能進行展示, 這個時候就可以用Zip了:

Compose操作符是將源Observable按照自定義的方式轉化成另外一個新的Observable。可以這么說compose是對Observable進行操作的而lift是對Subscriber進行操作的，作用點是不同的。
(這里的lift類似map，將數(shù)據(jù)類型轉化)

• fliter
  只發(fā)射通過了謂詞測試的數(shù)據(jù)項

filter.png