目錄

RxJava 到底是什麼？

非同步。

RxJava 的好處？

簡潔。隨著程式邏輯越來越複雜，它依然保持簡潔。

進入主題

基本實現

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!");
    }
    };

除了 Observer 介面之外，RxJava 還內建了一個實現了 Observer 的抽象類：Subscriber。 Subscriber 對 Observer 介面進行了一些擴充套件，但他們的基本使用方式是完全一樣的

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!");
    }
    };

例子：
由指定的一個 drawable 檔案 id drawableRes 取得圖片，並顯示在 ImageView 中，並在出現異常的時候列印 Toast 報錯：

int drawableRes = ...;
    ImageView imageView = ...;
    Observable.create(new OnSubscribe<Drawable>() {
    @Override
    public void call(Subscriber<? super Drawable> subscriber) {
    Drawable drawable = getTheme().getDrawable(drawableRes));
    subscriber.onNext(drawable);
    subscriber.onCompleted();
    }
    }).subscribe(new Observer<Drawable>() {
    @Override
    public void onNext(Drawable drawable) {
    imageView.setImageDrawable(drawable);
    }

    @Override
    public void onCompleted() {
        }

    @Override
    public void onError(Throwable e) {
    Toast.makeText(activity, "Error!", Toast.LENGTH_SHORT).show();
    }
    });

2).建立Observable

RxJava 使用 create() 方法來建立一個 Observable ，併為它定義事件觸發規則

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();
    }
    });

傳入了一個 OnSubscribe 物件作為引數。OnSubscribe 會被儲存在返回的 Observable 物件中，它的作用相當於一個計劃表，當 Observable 被訂閱的時候，OnSubscribe 的 call() 方法會自動被呼叫，事件序列就會依照設定依次觸發（對於上面的程式碼，就是觀察者Subscriber 將會被呼叫三次 onNext() 和一次 onCompleted()）。這樣，由被觀察者呼叫了觀察者的回撥方法，就實現了由被觀察者向觀察者的事件傳遞，即觀察者模式

RxJava 還提供了一些方法用來快捷建立事件佇列，例如：

1、 ————just(T…): 將傳入的引數依次傳送出來。

    Observable observable = Observable.just("Hello", "Hi", "Aloha");
// 將會依次呼叫：
// onNext("Hello");
// onNext("Hi");
// onNext("Aloha");
// onCompleted();

2、————from(T[]) / from(Iterable

String[] words = {"Hello", "Hi", "Aloha"};
Observable observable = Observable.from(words);
// 將會依次呼叫：
// onNext("Hello");
// onNext("Hi");
// onNext("Aloha");
// onCompleted();

3).Subscribe (訂閱)

建了 Observable 和 Observer 之後，再用 subscribe() 方法將它們聯結起來，整條鏈子就可以工作了

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

Observable.subscribe(Subscriber) 的內部實現是這樣的（僅核心程式碼）：

public Subscription subscribe(Subscriber subscriber) {
    subscriber.onStart();
    onSubscribe.call(subscriber);
        return subscriber;
}

可以看到，subscriber() 做了3件事：
1): 呼叫 Subscriber.onStart() 。這個方法在前面已經介紹過，是一個可選的準備方法。
2): 呼叫 Observable 中的 OnSubscribe.call(Subscriber) 。在這裡，事件傳送的邏輯開始執行。從這也可以看出，在 RxJava 中， Observable 並不是在建立的時候就立即開始傳送事件，而是在它被訂閱的時候，即當 subscribe() 方法執行的時候。 (訂閱的時候才開始傳送事件)。
3): 將傳入的 Subscriber 作為 Subscription 返回。這是為了方便 unsubscribe().

subscribe() 還支援不完整定義的回撥，RxJava 會自動根據定義創建出 Subscriber 。形式如下：

Action1<String> onNextAction = new Action1<String>() {
    // onNext()
    @Override
    public void call(String s) {
    Log.d(tag, s);
    }
    };
    Action1<Throwable> onErrorAction = new Action1<Throwable>() {
    // onError()
    @Override
    public void call(Throwable throwable) {
    // Error handling
    }
    };
    Action0 onCompletedAction = new Action0() {
    // onCompleted()
    @Override
    public void call() {
    Log.d(tag, "completed");
    }
    };
// 自動建立 Subscriber ，並使用 onNextAction 來定義 onNext()
observable.subscribe(onNextAction);
// 自動建立 Subscriber ，並使用 onNextAction 和 onErrorAction 來定義 onNext() 和 onError()
observable.subscribe(onNextAction, onErrorAction);
// 自動建立 Subscriber ，並使用 onNextAction、 onErrorAction 和 onCompletedAction 來定義 onNext()、 onError() 和 onCompleted()
observable.subscribe(onNextAction, onErrorAction, onCompletedAction);

Observer 和 Subscriber 具有相同的角色，而且 Observer 在 subscribe() 過程中最終會被轉換成 Subscriber 物件

Action0 是 RxJava 的一個介面，它只有一個方法 call()，這個方法是無參無返回值的；由於 onCompleted() 方法也是無參無返回值的，因此 Action0 可以被當成一個包裝物件，將 onCompleted() 的內容打包起來將自己作為一個引數傳入 subscribe() 以實現不完整定義的回撥，這樣其實也可以看做將 onCompleted() 方法作為引數傳進了 subscribe()

Action1 也是一個介面，它同樣只有一個方法 call(T param)，這個方法也無返回值，但有一個引數；與 Action0 同理，由於 onNext(T obj) 和 onError(Throwable error) 也是單引數無返回值的，因此 Action1 可以將 onNext(obj) 和 onError(error) 打包起來傳入 subscribe() 以實現不完整定義的回撥

例子：
將字串陣列 names 中的所有字串依次打印出來：

    String[] names = ...;
    Observable.from(names)
    .subscribe(new Action1<String>() {
    @Override
    public void call(String name) {
    Log.d(tag, name);
    }
    });

分水嶺

1. 在 RxJava 的預設規則中，事件的發出和消費都是在同一個執行緒的。 觀察者模式本身的目的就是『後臺處理，前臺回撥』的非同步機制，因此，則需要用到 RxJava 的另一個概念： Scheduler ，來實現非同步。
2. 在不指定執行緒的情況下， RxJava 遵循的是執行緒不變的原則，即：在哪個執行緒呼叫 subscribe()，就在哪個執行緒生產事件；在哪個執行緒生產事件，就在哪個執行緒消費事件。

Scheduler

Scheduler （執行緒控制）

RxJava 已經內建了幾個 Scheduler：
Schedulers.immediate() 預設的，直接在當前執行緒執行。
Schedulers.newThread() 啟用新執行緒，在新執行緒工作。
Schedulers.io() I/O操作（讀寫檔案，讀寫資料庫，網路資訊互動等）、和newThread（）最大的區別是：io() 的內部實現是一個無數量上限的執行緒池，可以重用空閒的執行緒。
AndroidSchedulers.mainThread() Android專用的，指定的操作在Android的主執行緒執行。

使用方法：

1. subscribeOn() 指定subscribe() 所發生的執行緒，即 Observable.OnSubcribe 被啟用時所處的執行緒，或者叫事件產生的執行緒。

2. observerOn() 指定Subscriber 執行所在的執行緒，或者叫做事件消費的執行緒。

文字敘述總歸難理解，上程式碼：

Observable.just(1, 2, 3, 4)
    .subscribeOn(Schedulers.io()) // 指定 subscribe() 發生在 IO 執行緒
    .observeOn(AndroidSchedulers.mainThread()) // 指定 Subscriber 的回調發生在主執行緒
    .subscribe(new Action1<Integer>() {
    @Override
    public void call(Integer number) {
    Log.d(tag, "number:" + number);
    }
    });

上面這段程式碼中，由於 subscribeOn(Schedulers.io()) 的指定，被建立的事件的內容 1、2、3、4 將會在 IO 執行緒發出；

而由於observeOn(AndroidScheculers.mainThread()) 的指定，因此subscriber 數字的列印將發生在主執行緒 。

事實上，這種在subscribe() 之前寫上兩句subscribeOn(Scheduler.io()) 和observeOn(AndroidSchedulers.mainThread()) 的使用方式非常常見，它適用於多數的 『後臺執行緒取資料，主執行緒顯示』的程式策略。

再看下面這個：

int drawableRes = ...;
    ImageView imageView = ...;
    Observable.create(new OnSubscribe<Drawable>() {
    @Override
    public void call(Subscriber<? super Drawable> subscriber) {
    Drawable drawable = getTheme().getDrawable(drawableRes));
    subscriber.onNext(drawable);
    subscriber.onCompleted();
    }
    })
    .subscribeOn(Schedulers.io()) // 指定 subscribe() 發生在 IO 執行緒
    .observeOn(AndroidSchedulers.mainThread()) // 指定 Subscriber 的回調發生在主執行緒
    .subscribe(new Observer<Drawable>() {
    @Override
    public void onNext(Drawable drawable) {
    imageView.setImageDrawable(drawable);
    }

    @Override
    public void onCompleted() {
    }

    @Override
    public void onError(Throwable e) {
    Toast.makeText(activity, "Error!", Toast.LENGTH_SHORT).show();
    }
    });

那麼，載入圖片將會發生在 IO 執行緒，而設定圖片則被設定在了主執行緒。

理解記憶過程

observerOn() 指定的是觀察者（警察）的執行緒

subscribeOn() 指定的是被觀察者（小偷）的執行緒

 比如上面的載入圖片的例子，在建立警察的時候，newObserver<Drawable>()，就定義好了，警察該乾的事，就是setImageDrawable(drawable)，設定圖片，所以上述例子就是，指定了警察在主執行緒設定圖片。而，在定義小偷的時候，Observable.create，就定了小偷會幹什麼事，就是獲得載入圖片的資源，那麼上訴例子就是，指定了小偷在IO執行緒，載入圖片.

這樣，一抽像一下理解，就加深理解多了。

變換

所謂變換，就是將事件序列中的物件或整個序列進行加工處理，轉換成不同的事件或事件序列

map

例子1

Observable.just("images/logo.png") // 輸入型別 String
    .map(new Func1<String, Bitmap>() {
    @Override
    public Bitmap call(String filePath) { // 引數型別 String
    return getBitmapFromPath(filePath); // 返回型別 Bitmap
    }
    })
    .subscribe(new Action1<Bitmap>() {
    @Override
    public void call(Bitmap bitmap) { // 引數型別 Bitmap
    showBitmap(bitmap);
    }
    });

Func1,它和 Action1 非常相似，也是 RxJava 的一個介面，用於包裝含有一個引數的方法。 Func1 和 Action 的區別在於， Func1 包裝的是有返回值的方法。

可以看到，map() 方法將引數中的 String 物件轉換成一個 Bitmap 物件後返回，而在經過 map() 方法後，事件的引數型別也由 String 轉為了 Bitmap。

它不僅可以針對事件物件，還可以針對整個事件佇列。

flatMap

首先假設這麼一種需求：假設有一個數據結構『學生』，現在需要打印出一組學生的名字。實現方式很簡單

private Students s1 = new Students("趙日天");
private Students s2 = new Students("陳鐵柱");
private Students s3 = new Students("王尼瑪");
private ArrayList<Students> list = new ArrayList<MainActivity.Students>();

    @Override
    protected void onCreate(Bundle savedInstanceState) {
    super.onCreate(savedInstanceState);
    setContentView(R.layout.activity_main);
    list.add(s1);
    list.add(s2);
    list.add(s3);

    Observable.from(list).map(new Func1<Students, String>() {

    @Override
    public String call(Students stu) {
    // TODO Auto-generated method stub
    return stu.getName();
    }
    }).subscribe(new Action1<String>() {

    @Override
    public void call(String s) {
    System.out.println("Students name is :" + s);

    }
    });

    }

    class Students {
    private String name;

    public Students(String name) {
    this.name = name;
    }

    public String getName() {
    return name;
    }

    public void setName(String name) {
    this.name = name;
    }

    }

執行結果：

那麼再假設：如果要打印出每個學生所需要修的所有課程的名稱呢？（需求的區別在於，每個學生只有一個名字，但卻有多個課程。）首先可以這樣實現

private List<Course> courseList = new ArrayList<MainActivity.Course>();
private Course c1 = new Course("紅Buff");
private Course c2 = new Course("藍Buff");
private Course c3 = new Course("火龍");
private Students s1;
private Students s2;
private Students s3;
private ArrayList<Students> list = new ArrayList<MainActivity.Students>();

    @Override
    protected void onCreate(Bundle savedInstanceState) {
    super.onCreate(savedInstanceState);
    setContentView(R.layout.activity_main);
    courseList.add(c1);
    courseList.add(c2);
    courseList.add(c3);
    s1 = new Students("趙日天", courseList);
    s2 = new Students("陳鐵柱", courseList);
    s3 = new Students("王尼瑪", courseList);
    list.add(s1);
    list.add(s2);
    list.add(s3);
    Observable.from(list).subscribe(new Subscriber<Students>() {

    @Override
    public void onCompleted() {
    // TODO Auto-generated method stub

    }

    @Override
    public void onError(Throwable arg0) {
    // TODO Auto-generated method stub

    }

    @Override
    public void onNext(Students student) {
    // TODO Auto-generated method stub
    List<Course> courses = student.getCourseList();
            for (int i = 0; i < courses.size(); i++) {
    Course course = courses.get(i);
    System.out.println("Student " + student.getName()
    + "Course name is :" + course.getcourseName());
    }

        }
    });

    }

    // 模擬學生結構
    class Students {
    private String name;
    private List<Course> courseList;

    public Students(String name, List<Course> courseList) {
        this.name = name;
        this.courseList = courseList;
    }

    public String getName() {
    return name;
    }

    public void setName(String name) {
    this.name = name;
    }

    public List<Course> getCourseList() {
    return courseList;
    }

    public void setCourseList(List<Course> courseList) {
    this.courseList = courseList;
    }
    }

    // 模擬課程資料結構
    class Course {
    private String courseName;

    public Course(String courseName) {
    this.courseName = courseName;
    }

    public String getcourseName() {
    return courseName;
    }

    public void setcourseName(String name) {
    this.courseName = name;
    }

    }

執行結果：

程式碼很簡單。但是，如果不想在 Subscriber 中使用 for 迴圈呢，而是希望 Subscriber 中直接傳入單個的 Course 物件呢？因為map() 是一對一的轉化，而現在的要求是一對多的轉化。那怎麼才能把一個 Student 轉化成多個 Course 呢？

這個時候，就需要用 flatMap() 了

flatMap

private List<Course> courseList = new ArrayList<MainActivity.Course>();
private Course c1 = new Course("紅Buff");
private Course c2 = new Course("藍Buff");
private Course c3 = new Course("火龍");
private Students s1;
private Students s2;
private Students s3;
private ArrayList<Students> list = new ArrayList<MainActivity.Students>();
    @Override
    protected void onCreate(Bundle savedInstanceState) {
    super.onCreate(savedInstanceState);
    setContentView(R.layout.activity_main);
    courseList.add(c1);
    courseList.add(c2);
    courseList.add(c3);
    s1 = new Students("趙日天", courseList);
    s2 = new Students("陳鐵柱", courseList);
    s3 = new Students("王尼瑪", courseList);
    list.add(s1);
    list.add(s2);
    list.add(s3);
    Observable.from(list)
    .flatMap(new Func1<Students, Observable<Course>>() {
    @Override
    public Observable<Course> call(Students stu) {
    // TODO Auto-generated method stub
    System.out.println("學生" + stu.getName() + "的課程如下：");
                    return Observable.from(stu.getCourseList());
    }
            }).subscribe(new Action1<Course>() {
    @Override
    public void call(Course cou) {
    System.out.println(cou.getcourseName());
    }
        });
    }
    // 模擬學生結構
    class Students {
    private String name;
    private List<Course> courseList;
    public Students(String name, List<Course> courseList) {
        this.name = name;
        this.courseList = courseList;
    }
    public String getName() {
    return name;
    }
    public void setName(String name) {
    this.name = name;
    }
    public List<Course> getCourseList() {
    return courseList;
    }
    public void setCourseList(List<Course> courseList) {
    this.courseList = courseList;
    }
    }
    // 模擬課程資料結構
    class Course {
    private String courseName;
    public Course(String courseName) {
    this.courseName = courseName;
    }
    public String getcourseName() {
    return courseName;
    }
    public void setcourseName(String name) {
    this.courseName = name;
    }
    }

執行結果：

可以看出， flatMap() 和 map() 有一個相同點：它也是把傳入的引數轉化之後返回另一個物件。但需要注意，和 map() 不同的是， flatMap() 中返回的是個 Observable 物件，並且這個 Observable 物件並不是被直接傳送到了 Subscriber 的回撥方法中。

flatMap() 的原理

1. 使用傳入的事件物件建立一個 Observable 物件；
2. 並不傳送這個 Observable, 而是將它啟用，於是它開始傳送事件；
3. 每一個創建出來的 Observable 傳送的事件，都被匯入同一個 Observable,而這個 Observable 負責將這些事件統一交給 Subscriber 的回撥方法。

個人理解

 特工（被觀察者）的能力是給他傳入一堆學生，他先自己查到這個學生的全部課程，由於每個學生的課程都很多，所以不能直接讓監視他的上級知道，他自己有一個情報小組，把課程給到情報小組，然後整理好情報，最後由情報小組統一再給到時刻關注監視他的上級（觀察者）。在該事件中，特工把事件拆成了兩級，通過一組新建立的 Observable 將初始的物件『鋪平』之後通過統一路徑分發了下去。也就是，把每一個學生的每一項課程，鋪平後，按順序的列好告訴上級。

說點什麼吧

終於到這了，說句實話，扔物線的那篇文章是真的棒，我看了三遍才能看得懂，第一遍看的時候，整個人都是懵逼狀態的，天啊，這什麼啊

可是，必須硬著頭皮看下去啊，看第二遍的時候，就感覺很舒服了，看到執行結果的時候，真的棒棒大。(⊙o⊙)…

該工程包括 ：

1、MvpDemo （一個簡單的MVP架構的demo）

2、RetrofitRxJavaDemo (一個使用RxJava + Retrofit 的小demo)

至於RxJava 和 Retrofit ，看的是拋物線的文章，還有別的一些。這裡有Fork了一個挺好的蒐集，不過，個人認為，有很多選擇看就好了，有些不一定好。 ：https://github.com/GitHuborz/Awesome-RxJava

最近的計劃是，學完這些後，寫一個用Gank.io的介面，基於 RxJava+Retrofit+MVP+OkHttp，當然也要使用時下流行許久了的Rv+MD等等的一個妹子APP。（哈哈）

總之，一步一步實現吧，Go~

想看妹子，請看下回分解~