Rxjava復(fù)習專用

Rxjava —— 四個基本概念

• RxJava 有四個基本概念：Observable (可觀察者，即被觀察者)、 Observer (觀察者)、 subscribe (訂閱)、事件。
• Observable 和 Observer 通過 subscribe() 方法實現(xiàn)訂閱關(guān)系，從而 Observable 可以在需要的時候發(fā)出事件來通知 Observer。

Observer —— 觀察者

它決定事件觸發(fā)的時候?qū)⒂性鯓拥男袨?/h6>

• Observer 接口
• Subscriber 抽象類,實現(xiàn)了Observer和Subscription
  不僅基本使用方式一樣，實質(zhì)上，在 RxJava 的 subscribe 過程中，Observer 也總是會先被轉(zhuǎn)換成一個 Subscriber 再使用。所以如果你只想使用基本功能，選擇 Observer 和 Subscriber 是完全一樣的。它們的區(qū)別對于使用者來說主要有兩點：

兩者的區(qū)別:
1.Subscriber 多了 onStart( ) 方法,但是該方法發(fā)生在 subscribe 發(fā)生的線程,不能指定線程
2.Subscriber 多了 unsubscribe( ) 方法, Subscriber 所實現(xiàn)的另一個接口 Subscription 的方法，用于取消訂閱

示例代碼:

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

Observable —— 被觀察者

它決定什么時候觸發(fā)事件以及觸發(fā)怎樣的事件

示例代碼:

• create( ) RxJava 最基本的創(chuàng)造事件序列的方法

Observable observable = Observable.create(new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> subscriber) {
                //觀察者將會依次調(diào)用三次onNext和一次Completed
                subscriber.onNext("Hello");
                subscriber.onNext("Hi");
                subscriber.onNext("Aloha");
                subscriber.onCompleted();
            }
        });

• just( T... ) 將傳入的參數(shù)依次發(fā)送出來。

Observable observable = Observable.just("Hello", "Hi", "Aloha");
// 將會依次調(diào)用：
// onNext("Hello");
// onNext("Hi");
// onNext("Aloha");
// onCompleted();

• from(T[ ])

String[] words = {"Hello", "Hi", "Aloha"};
Observable observable = Observable.from(words);
// 將會依次調(diào)用：
// onNext("Hello");
// onNext("Hi");
// onNext("Aloha");
// onCompleted();

Subscribe —— 訂閱

創(chuàng)建了 Observable 和 Observer 之后，再用 subscribe() 方法將它們聯(lián)結(jié)起來，整條鏈子就可以工作了

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

Observable.subscribe(Subscriber) 的內(nèi)部實現(xiàn)是這樣的（僅核心代碼）：

// 注意：這不是 subscribe() 的源碼，而是將源碼中與性能、兼容性、擴展性有關(guān)的代碼剔除后的核心代碼。
// 如果需要看源碼，可以去 RxJava 的 GitHub 倉庫下載。
public Subscription subscribe(Subscriber subscriber) {
    subscriber.onStart();
    onSubscribe.call(subscriber);
    return subscriber;
}

調(diào)用 Subscriber.onStart() 。這個方法在前面已經(jīng)介紹過，是一個可選的準備方法。
調(diào)用 Observable 中的 OnSubscribe.call(Subscriber) 。在這里，事件發(fā)送的邏輯開始運行。從這也可以看出，在 RxJava 中， Observable 并不是在創(chuàng)建的時候就立即開始發(fā)送事件，而是在它被訂閱的時候，即當 subscribe() 方法執(zhí)行的時候。
將傳入的 Subscriber 作為 Subscription 返回。這是為了方便 unsubscribe().

除了 subscribe(Observer) 和 subscribe(Subscriber) ，subscribe() 還支持不完整定義的回調(diào)，RxJava 會自動根據(jù)定義創(chuàng)建出 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");
    }
};

// 自動創(chuàng)建 Subscriber ，并使用 onNextAction 來定義 onNext()
observable.subscribe(onNextAction);
// 自動創(chuàng)建 Subscriber ，并使用 onNextAction 和 onErrorAction 來定義 onNext() 和 onError()
observable.subscribe(onNextAction, onErrorAction);
// 自動創(chuàng)建 Subscriber ，并使用 onNextAction、 onErrorAction 和 onCompletedAction 來定義 onNext()、 onError() 和 onCompleted()
observable.subscribe(onNextAction, onErrorAction, onCompletedAction);

下面看兩個例子

• 打印字符串數(shù)組

將字符串數(shù)組 names 中的所有字符串依次打印出來：
String[] names = ...;
Observable.from(names)
    .subscribe(new Action1<String>() {
        @Override
        public void call(String name) {
            Log.d(tag, name);
        }
    });

• 由 id 取得圖片并顯示

由指定的一個 drawable 文件 id drawableRes 取得圖片，并顯示在 ImageView 中，并在出現(xiàn)異常的時候打印 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();
    }
});

以上為RxJava的基本使用,不涉及線程切換.以下開始切換

線程控制 —— Scheduler

相當于線程控制器，RxJava 通過它來指定每一段代碼應(yīng)該運行在什么樣的線程

• Scheduler的API

1. Schedulers.immediate(): 直接在當前線程運行，相當于不指定線程。這是默認的 Scheduler。
2. Schedulers.newThread(): 總是啟用新線程，并在新線程執(zhí)行操作。
3. Schedulers.io(): I/O 操作（讀寫文件、讀寫數(shù)據(jù)庫、網(wǎng)絡(luò)信息交互等）所使用的 Scheduler。行為模式和 newThread() 差不多，區(qū)別在于 io() 的內(nèi)部實現(xiàn)是是用一個無數(shù)量上限的線程池，可以重用空閑的線程，因此多數(shù)情況下 io() 比 newThread() 更有效率。不要把計算工作放在 io() 中，可以避免創(chuàng)建不必要的線程。
4. Schedulers.computation(): 計算所使用的 Scheduler。這個計算指的是 CPU 密集型計算，即不會被 I/O 等操作限制性能的操作，例如圖形的計算。這個 Scheduler 使用的固定的線程池，大小為 CPU 核數(shù)。不要把 I/O 操作放在 computation() 中，否則 I/O 操作的等待時間會浪費 CPU。
5. 另外， Android 還有一個專用的 AndroidSchedulers.mainThread()，它指定的操作將在 Android 主線程運行。

有了這幾個 Scheduler，就可以使用

• subscribeOn( )
• observeOn( )
  兩個方法來對線程進行控制了。subscribeOn( ) : 指定 subscribe() 所發(fā)生的線程，即 Observable.OnSubscribe 被激活時所處的線程?；蛘呓凶鍪录a(chǎn)生的線程。observeOn( ) : 指定 Subscriber 所運行在的線程。或者叫做事件消費的線程。

示例代碼:非常常見，它適用于多數(shù)的 『后臺線程取數(shù)據(jù)，主線程顯示』的程序策略。

//將圖片路徑轉(zhuǎn)為bitmap
Observable.just(1, 2, 3, 4)
    .subscribeOn(Schedulers.io()) // 指定 subscribe() 發(fā)生在 IO 線程
    .observeOn(AndroidSchedulers.mainThread()) // 指定 Subscriber 的回調(diào)發(fā)生在主線程
    .subscribe(new Action1<Integer>() {
        @Override
        public void call(Integer number) {
            Log.d(tag, "number:" + number);
        }
    });

變換 —— 將事件序列中的對象或整個序列進行加工處理，轉(zhuǎn)換成不同的事件或事件序列

API

• map( ) 事件對象的一對一的變換

//將圖片路徑轉(zhuǎn)為bitmap
Observable.just("images/logo.png") // 輸入類型 String
    .map(new Func1<String, Bitmap>() {
        @Override
            public Bitmap call(String filePath) { // 參數(shù)類型 String
                return getBitmapFromPath(filePath); // 返回類型 Bitmap
            }
        })
    .subscribe(new Action1<Bitmap>() {
        @Override
        public void call(Bitmap bitmap) { // 參數(shù)類型 Bitmap
             showBitmap(bitmap);
        }
     });

• flatMap( ) 一堆對象變成另一堆對象

//假設(shè)有一個數(shù)據(jù)結(jié)構(gòu)『學生』，現(xiàn)在需要打印出每個學生所需要修的所有課程的名稱
Student[] students = ...;
Subscriber<Course> subscriber = new Subscriber<Course>() {
   @Override
    public void onNext(Course course) {
        Log.d(tag, course.getName());
    }
    ...
};
Observable.from(students)
    .flatMap(new Func1<Student, Observable<Course>>() {
        @Override
        public Observable<Course> call(Student student) {
            return Observable.from(student.getCourses());//這句代碼
        }
    })
    .subscribe(subscriber);

解析:

將一堆對象拆分,變成一個個對象,再以一個對象為單位創(chuàng)建觀察者,Observable.from(student.getCourses())這句代碼看出,新建的觀察者發(fā)送事件.全部匯入Subscriber的回調(diào)中

• flatMap() 和 map() 有一個相同點：它也是把傳入的參數(shù)轉(zhuǎn)化之后返回另一個對象。
• 但需要注意，和 map() 不同的是， flatMap() 中返回的是個 Observable(觀察者) 對象，并且這個 Observable 對象并不是被直接發(fā)送到了 Subscriber 的回調(diào)方法中。
• flatMap() 的原理是這樣的：
  1. 使用傳入的事件對象創(chuàng)建一個 Observable 對象；
  2. 并不發(fā)送這個 Observable, 而是將它激活，于是它開始發(fā)送事件；
  3. 每一個創(chuàng)建出來的 Observable 發(fā)送的事件，都被匯入同一個 Observable ，而這個 Observable 負責將這些事件統(tǒng)一交給 Subscriber 的回調(diào)方法。這三個步驟，把事件拆成了兩級，通過一組新創(chuàng)建的 Observable 將初始的對象『鋪平』之后通過統(tǒng)一路徑分發(fā)了下去。而這個『鋪平』就是 flatMap() 所謂的 flat。

這里略過了變換的原理

• compose( ) 一堆被觀察者都需要變化相同的邏輯,使用此方法

 //假設(shè)在程序中有多個 Observable ，并且他們都需要應(yīng)用一組相同的 lift() 變換。
public class LiftAllTransformer implements Observable.Transformer<Integer, String> {
    @Override
    public Observable<String> call(Observable<Integer> observable) {
        return observable
            .lift1()
            .lift2()
            .lift3()
            .lift4();
    }
}
...
Transformer liftAll = new LiftAllTransformer();
observable1.compose(liftAll).subscribe(subscriber1);
observable2.compose(liftAll).subscribe(subscriber2);
observable3.compose(liftAll).subscribe(subscriber3);
observable4.compose(liftAll).subscribe(subscriber4);

線程控制 —— 線程的自由控制

代碼示例

Observable.just(1, 2, 3, 4) // IO 線程，由 subscribeOn() 指定
    .subscribeOn(Schedulers.io())//===指定了事件發(fā)生的線程. subscribeOn() 的位置放在哪里都可以，但它是只能調(diào)用一次的。
    .observeOn(Schedulers.newThread())//================指定了接下來map發(fā)送的線程
    .map(mapOperator) // 新線程，由 observeOn() 指定
    .observeOn(Schedulers.io())//=======================指定了接下來map發(fā)送的線程
    .map(mapOperator2) // IO 線程，由 observeOn() 指定
    .observeOn(AndroidSchedulers.mainThread) //==========制定了接下來消費事件的線程
    .subscribe(subscriber);  // Android 主線程，由 observeOn() 指定

解析:
observeOn() 指定的是 Subscriber(觀察者) 的線程，而這個 Subscriber(觀察者) 并不是（嚴格說應(yīng)該為『不一定是』，但這里不妨理解為『不是』）subscribe() 參數(shù)中的 Subscriber(觀察者)，而是 observeOn() 執(zhí)行時的當前 Observable 所對應(yīng)的 Subscriber(觀察者)，即它的直接下級 Subscriber(觀察者) 。換句話說，observeOn() 指定的是它之后的操作所在的線程。因此如果有多次切換線程的需求，只要在每個想要切換線程的位置調(diào)用一次 observeOn() 即可
意思就是:RxJava中對Observable進行變化操作都會有對應(yīng)的Subscriber(觀察者) ,我們在操作前設(shè)置線程observeOn()就可以指定接下來的Subscriber(觀察者)

延伸：doOnSubscribe()

Observable.create(onSubscribe)
    .subscribeOn(Schedulers.io())
    .doOnSubscribe(new Action0() {
        @Override
        public void call() {
            progressBar.setVisibility(View.VISIBLE); // 需要在主線程執(zhí)行
        }
    })
    .subscribeOn(AndroidSchedulers.mainThread()) // 指定主線程,
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(subscriber);

解析:
它和 Subscriber.onStart() 同樣是在 subscribe() 調(diào)用后而且在事件發(fā)送前執(zhí)行，但區(qū)別在于它可以指定線程。默認情況下， doOnSubscribe() 執(zhí)行在 subscribe() 發(fā)生的線程；而如果在 doOnSubscribe() 之后有 subscribeOn() 的話，它將執(zhí)行在離它最近的 subscribeOn() 所指定的線程。

差不多了,該去實際場景中使用了

復(fù)習鏈接
給 Android 開發(fā)者的 RxJava 詳解 -- [作者：扔物線]