Rxjava2 Observable的輔助操作詳解及實例（一）

目錄

• 簡要：
• 1. Delay
• 2. Do
• 3. SubscribeOn
• 4. ObserverOn
• 5. Serialize
• 6. Materialize
• 7. Dematerialize
• 接續：

簡要：

需求了解：

Rxjava中有一些方便的輔助操作符，來更方便我們的函數式的編程。比如延遲、定時、指定操作的監聽、數據類型轉換等一系列的操作。

下面列出了一些用於Observable的輔助操作符：

• Delay：延時發射Observable的結果。
• Do：注冊一個動作作為原始Observable生命周期事件的監聽器。
• SubscribeOn：指定Observable自身在哪個調度器上執行。
• ObserverOn：指定一個觀察者在哪個調度器上觀察這個Observable。
• Serialize：強制一個Observable連續調用並保證行為正確，其實就是同步事件操作。
• Materialize/Dematerialize：將數據項和事件通知都當做數據項發射 ，Dematerialize 剛好相反。
• TimeInterval：將一個發射數據的Observable轉換為發射那些數據發射時間間隔的Observable。
• Timeout：對原始Observable的一個鏡像，如果過了一個指定的時長仍沒有發射數據，它會發一個錯誤通知。
• Timestamp：給Observable發射的數據項附加一個指定的時間戳。
• Using：創建一個只在Observable生命周期內存在的一次性資源。
• To：將Observable轉換為另一個對象或數據結構。

1. Delay

延遲一段指定的時間再發射來自Observable的發射物。

Delay 操作符讓原始 Observable 在發射每項數據之前都暫停一段指定的時間段。效果是Observable發射的數據項在時間上向前整體平移了一個增量。

1.1 delay(long delay, TimeUnit unit)

延遲指定時間段后發射原始Observable發射的數據序列，如果發生異常的話，會立即發射通知給觀察者。

1.2 delay(Function<T, ObservableSource> itemDelay)

使用一個函數針對原始 Observable 的每一項數據返回一個 Observable ，它監視返回的這個 Observable ，當任何那樣的 Observable 終止時，delay 返回的 Observable 就發射關聯的那項數據。

1.3 delay(ObservableSource subscriptionDelay, Function<T, ObservableSource> itemDelay)

延遲直到 subscriptionDelay 發射第一個數據項后開始訂閱原始 Observable，然后再使用一個函數針對原始Observable的每一項數據返回一個Observable，它監視返回的這個Observable，當任何那樣的 Observable 終止時，delay 返回的 Observable 就發射關聯的那項數據。

示例代碼：

    // 創建Observable
    Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
        @Override
        public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
            emitter.onNext(1);
            emitter.onNext(2);
            emitter.onNext(3);
        //  emitter.onError(new Exception("Test Error!"));
            emitter.onNext(4);
            emitter.onNext(5);
            emitter.onComplete();
        }
    });

    /**
     * 1. delay(long delay, TimeUnit unit,
     *  Scheduler scheduler: 可選參數，指定工作線程
     *  boolean delayError:    可選參數，延遲異常通知到最后
     *  )
     *  延遲指定時間段后發射原始Observable發射的數據序列，如果發生異常的話，會立即發射通知給觀察者。
     */
    observable.doOnNext(new Consumer<Integer>() {
        @Override
        public void accept(Integer integer) throws Exception {
            System.out.println("--> doOnNext(1): " + integer);
        }

    }).delay(1, TimeUnit.SECONDS, Schedulers.newThread(), false) // 在子線程中延遲1秒發射數據，不延遲異常通知
      .subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                System.out.println("--> onSubscribe(1)");
            }

            @Override
            public void onNext(Integer integer) {
                System.out.println("--> onNext(1): " + integer);
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("--> onError(1): " + e);
            }

            @Override
            public void onComplete() {
                System.out.println("--> onCompleted(1)");
            }
      });

    System.in.read();
    System.out.println("-----------------------------------------------------");
    /**
     *  2. delay(Function<T, ObservableSource<U>> itemDelay)
     *   使用一個函數針對原始Observable的每一項數據返回一個Observable，它監視返回的這個Observable，
     *   當任何那樣的 Observable 終止時，delay 返回的 Observable 就發射關聯的那項數據。
     */
    observable.doOnNext(new Consumer<Integer>() {
        @Override
        public void accept(Integer integer) throws Exception {
            System.out.println("--> doOnNext(2): " + integer);
        }

    }).delay(new Function<Integer, ObservableSource<Long>>() {
        @Override
        public ObservableSource<Long> apply(Integer integer) throws Exception {
            System.out.println("--> ObservableSource(2)： " + integer);
            Observable<Long> timer = Observable.timer(integer, TimeUnit.SECONDS);
            return timer;
        }
    }).subscribe(new Observer<Integer>() {
        @Override
        public void onSubscribe(Disposable d) {
            System.out.println("--> onSubscribe(2)");
        }

        @Override
        public void onNext(Integer integer) {
            System.out.println("--> onNext(2): " + integer);
        }

        @Override
        public void onError(Throwable e) {
            System.out.println("--> onError(2): " + e);
        }

        @Override
        public void onComplete() {
            System.out.println("--> onCompleted(2)");
        }
    });

    System.in.read();
    System.out.println("-----------------------------------------------------");
    /**
     *  3. delay(ObservableSource subscriptionDelay, Function<T, ObservableSource> itemDelay)
     *  延遲直到subscriptionDelay發射第一個數據項后開始訂閱原始Observable
     *  然后再使用一個函數針對原始Observable的每一項數據返回一個Observable，它監視返回的這個Observable，
     *  當任何那樣的 Observable 終止時，delay 返回的 Observable 就發射關聯的那項數據。
     */
    observable.doOnNext(new Consumer<Integer>() {
        @Override
        public void accept(Integer integer) throws Exception {
            System.out.println("--> doOnNext(3): " + integer);
        }
       // 延遲3秒后開始訂閱源Observable，然后對發射的每項數據進行function函數延遲
    }).delay(Observable.timer(3, TimeUnit.SECONDS), new Function<Integer, ObservableSource<Long>>() {
        @Override
        public ObservableSource<Long> apply(Integer integer) throws Exception {
            System.out.println("--> apply(3): " + integer);
            return Observable.timer(integer, TimeUnit.SECONDS);
        }
    }).subscribe(new Observer<Integer>() {
        @Override
        public void onSubscribe(Disposable d) {
            System.out.println("--> onSubscribe(3)");
        }

        @Override
        public void onNext(Integer integer) {
            System.out.println("--> onNext(3): " + integer);
        }

        @Override
        public void onError(Throwable e) {
            System.out.println("--> onError(3): " + e);
        }

        @Override
        public void onComplete() {
            System.out.println("--> onCompleted(3)");
        }
    });

    System.in.read();

輸出：

--> onSubscribe(1)
--> doOnNext(1): 1
--> doOnNext(1): 2
--> doOnNext(1): 3
--> doOnNext(1): 4
--> doOnNext(1): 5
--> onNext(1): 1
--> onNext(1): 2
--> onNext(1): 3
--> onNext(1): 4
--> onNext(1): 5
--> onCompleted(1)
-----------------------------------------------------
--> onSubscribe(2)
--> doOnNext(2): 1
--> ObservableSource(2)： 1
--> doOnNext(2): 2
--> ObservableSource(2)： 2
--> doOnNext(2): 3
--> ObservableSource(2)： 3
--> doOnNext(2): 4
--> ObservableSource(2)： 4
--> doOnNext(2): 5
--> ObservableSource(2)： 5
--> onNext(2): 1
--> onNext(2): 2
--> onNext(2): 3
--> onNext(2): 4
--> onNext(2): 5
--> onCompleted(2)
-----------------------------------------------------
--> onSubscribe(3)
--> doOnNext(3): 1
--> apply(3): 1
--> doOnNext(3): 2
--> apply(3): 2
--> doOnNext(3): 3
--> apply(3): 3
--> doOnNext(3): 4
--> apply(3): 4
--> doOnNext(3): 5
--> apply(3): 5
--> onNext(3): 1
--> onNext(3): 2
--> onNext(3): 3
--> onNext(3): 4
--> onNext(3): 5
--> onCompleted(3)

Javadoc: delay(long delay, TimeUnit unit, Scheduler scheduler, boolean delayError)
Javadoc: delay(Function<T, ObservableSource> itemDelay)
Javadoc: delay(ObservableSource subscriptionDelay, Function<T, ObservableSource> itemDelay)

2. Do

注冊一個動作作為原始Observable生命周期事件的監聽器。

你可以注冊指定的回調，當Observable的某個事件發生時，Rxjava 會在與 Observable 鏈關聯的正常通知集合中調用它。

在Rxjava中有許多相關Do的變體，分別進行不同場景的事件監聽，一般有下面幾種操作方法：

1. doOnSubscribe(Consumer onSubscribe)：一旦有觀察者訂閱了Observable，就會被調用。
2. doOnLifecycle(Consumer onSubscribe, Action onDispose)： 在觀察者訂閱產生和解除時被調用。
3. doOnNext(Consumer onNext)：在 Observable 每次發射數據前被調用。
4. doOnEach(Observer observer): 在 Observable 調用觀察者的所有通知前被調用。
5. doAfterNext(Consumer onAfterNext)：在 Observable 調用OnNext通知(數據發射通知)之后被調用。
6. doOnError(Consumer onError)：注冊一個動作，當它的 Observable 由於異常終止調用 onError 時會被調用。
7. doOnTerminate(Action onTerminate): 當Observable終止之前會被調用，無論是正常還是異常終止。
8. doAfterTerminate(Action onFinally): 當Observable終止之后會被調用，無論是正常還是異常終止。
9. doOnComplete(Action onComplete)：Observable正常終止調用 onCompleted 時會被調用。
10. doFinally(Action onFinally)：Observable終止之后會被調用，無論是正常還是異常終止，但是優先於doAfterTerminate。
11. doOnDispose(Action onDispose)：在觀察者調用Disposable的dispose()方法時被調用。

示例代碼：

    /**
     *  1. doOnSubscribe(Consumer onSubscribe)
     *  一旦有觀察者訂閱了Observable，就會被調用
     */
    Observable.just(999).doOnSubscribe(new Consumer<Disposable>() {
        @Override
        public void accept(Disposable disposable) throws Exception {
            System.out.println("----> doOnSubscribe");
        }
    }).subscribe(new Observer<Integer>() {
        @Override
        public void onSubscribe(Disposable d) {
            System.out.println("--> onSubscribe(1)");
        }

        @Override
        public void onNext(Integer integer) {
            System.out.println("--> onNext(1): " + integer);
        }

        @Override
        public void onError(Throwable e) {
            System.out.println("--> onError(1): " + e);
        }

        @Override
        public void onComplete() {
            System.out.println("--> onCompleted(1)");
        }
    });

    System.out.println("--------------------------------------------");
    /**
     *  2. doOnLifecycle(Consumer onSubscribe, Action onDispose)
     *  onSubscribe： 接受觀察者訂閱前的通知，可以在此通知中解除訂閱
     *  onDispose：   接受觀察者調用解除訂閱通知
     *  在觀察者訂閱產生和解除時調用
     */
    Observable.just(999).doOnLifecycle(new Consumer<Disposable>() {
        @Override
        public void accept(Disposable disposable) throws Exception {
            System.out.println("----> doOnLifecycle onSubscribe(2)");
            // disposable.dispose();  // 可以在觀察者訂閱前直接解除訂閱
        }
    }, new Action() {
        @Override
        public void run() throws Exception {
            System.out.println("----> doOnLifecycle onDispose(2)");
        }
    }).subscribe(new Observer<Integer>() {
        private Disposable disposable;

        @Override
        public void onSubscribe(Disposable d) {
            disposable = d;
            System.out.println("--> onSubscribe(2)");
        }

        @Override
        public void onNext(Integer integer) {
            System.out.println("--> onNext(2): " + integer);
            disposable.dispose(); // 手動解除訂閱
        }

        @Override
        public void onError(Throwable e) {
            System.out.println("--> onError(2): " + e);
        }

        @Override
        public void onComplete() {
            System.out.println("--> onCompleted(2)");
        }
    });

    System.out.println("--------------------------------------------");
    /**
     *  3. doOnNext(Consumer onNext)
     *  在Observable每次發射數據前被調用
     */
    Observable.just(999).doOnNext(new Consumer<Integer>() {
        @Override
        public void accept(Integer integer) throws Exception {
            System.out.println("----> doOnNext(3): " + integer);
        }
    }).subscribe(new Consumer<Integer>() {
        @Override
        public void accept(Integer integer) throws Exception {
            System.out.println("--> accept(3): " + integer);
        }
    });

    System.out.println("--------------------------------------------");
    /**
     *  4. doOnEach(Observer observer)
     *  在Observable調用觀察者的所有通知前被調用
     */
    Observable.just(999).doOnEach(new Observer<Integer>() {
        @Override
        public void onSubscribe(Disposable d) {
            System.out.println("----> doOnEach(4) onSubscribe");
        }

        @Override
        public void onNext(Integer integer) {
            System.out.println("----> doOnEach(4) onNext: " + integer);
        }

        @Override
        public void onError(Throwable e) {
            System.out.println("----> doOnEach(4) onError: " + e);
        }

        @Override
        public void onComplete() {
            System.out.println("----> doOnEach(4) onComplete");
        }
    }).subscribe(new Observer<Integer>() {

        @Override
        public void onSubscribe(Disposable d) {
            System.out.println("--> onSubscribe(4)");
        }

        @Override
        public void onNext(Integer integer) {
            System.out.println("--> onNext(4): " + integer);
        }

        @Override
        public void onError(Throwable e) {
            System.out.println("--> onError(4): " + e);
        }

        @Override
        public void onComplete() {
            System.out.println("--> onCompleted(4)");
        }
    });

    System.out.println("--------------------------------------------");
    /**
     *  5. doAfterNext(Consumer onAfterNext)
     *  在Observable調用OnNext通知(數據發射通知)之后被調用
     */
    Observable.just(999).doAfterNext(new Consumer<Integer>() {
        @Override
        public void accept(Integer integer) throws Exception {
            System.out.println("----> doAfterNext(5): " + integer);
        }
    }).subscribe(new Consumer<Integer>() {
        @Override
        public void accept(Integer integer) throws Exception {
            System.out.println("--> onNext(5): " + integer);
        }
    });

    System.out.println("--------------------------------------------");
    /**
     *  6. doOnError(Consumer onError)
     *  注冊一個動作，當它的Observable由於異常終止調用 onError 時會被調用
     */
    Observable.create(new ObservableOnSubscribe<Integer>() {
        @Override
        public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
            emitter.onError(new Exception("Test Error!"));
        }
    }).doOnError(new Consumer<Throwable>() {
        @Override
        public void accept(Throwable throwable) throws Exception {
            System.out.println("----> doOnError(6): " + throwable);
        }
    }).subscribe(new Observer<Integer>() {
        @Override
        public void onSubscribe(Disposable d) {
            System.out.println("--> onSubscribe(6)");
        }

        @Override
        public void onNext(Integer integer) {
            System.out.println("--> onNext(6): " + integer);
        }

        @Override
        public void onError(Throwable e) {
            System.out.println("--> onError(6): " + e);
        }

        @Override
        public void onComplete() {
            System.out.println("--> onComplete(6)");
        }
    });

    System.out.println("--------------------------------------------");
    /**
     *  7.
     *  doOnTerminate(Action onTerminate):  當Observable終止之前會被調用，無論是正常還是異常終止
     *  doAfterTerminate(Action onFinally): 當Observable終止之后會被調用，無論是正常還是異常終止
     */
    Observable.just(999).doOnTerminate(new Action() {
        @Override
        public void run() throws Exception {
            System.out.println("----> doOnTerminate(7)");
        }
    }).subscribe(new Consumer<Integer>() {
        @Override
        public void accept(Integer integer) throws Exception {
            System.out.println("--> accept(7): " + integer);
        }
    });

    System.out.println("--------------------------------------------");
    /**
     *  8. doOnComplete(Action onComplete)
     *  Observable正常終止調用 onCompleted 時會被調用
     */
    Observable.just(999).doOnComplete(new Action() {
        @Override
        public void run() throws Exception {
            System.out.println("----> doOnComplete(8)");
        }
    }).subscribe(new Observer<Integer>() {
        @Override
        public void onSubscribe(Disposable d) {
            System.out.println("--> onSubscribe(8)");
        }

        @Override
        public void onNext(Integer integer) {
            System.out.println("--> onNext(8): " + integer);
        }

        @Override
        public void onError(Throwable e) {
            System.out.println("--> onError(8): " + e);
        }

        @Override
        public void onComplete() {
            System.out.println("--> onComplete(8)");
        }
    });

    System.out.println("--------------------------------------------");
    /**
     *  9. doFinally(Action onFinally)
     *  Observable終止之后會被調用，無論是正常還是異常終止，但是優先於doAfterTerminate
     */
    Observable.just(999).doFinally(new Action() {
        @Override
        public void run() throws Exception {
            System.out.println("----> doFinally(9)");
        }
    }).subscribe(new Observer<Integer>() {
        @Override
        public void onSubscribe(Disposable d) {
            System.out.println("--> onSubscribe(9)");
        }

        @Override
        public void onNext(Integer integer) {
            System.out.println("--> onNext(9): " + integer);
        }

        @Override
        public void onError(Throwable e) {
            System.out.println("--> onError(9): " + e);
        }

        @Override
        public void onComplete() {
            System.out.println("--> onComplete(9)");
        }
    });
	
    System.out.println("--------------------------------------------");
    /**
     *  10. doOnDispose(Action onDispose)
     *  在觀察者調用Disposable的dispose()方法時被調用
     */
    Observable.just(999).doOnDispose(new Action() {
        @Override
        public void run() throws Exception {
            System.out.println("----> doOnDispose(10)");
        }
    }).subscribe(new Observer<Integer>() {
        @Override
        public void onSubscribe(Disposable d) {
            System.out.println("--> onSubscribe(10)");
            d.dispose();
        }

        @Override
        public void onNext(Integer integer) {
            System.out.println("--> onNext(10): " + integer);
        }

        @Override
        public void onError(Throwable e) {
            System.out.println("--> onError(10): " + e);
        }

        @Override
        public void onComplete() {
            System.out.println("--> onComplete(10)");
        }
    });

輸出：

----> doOnSubscribe
--> onSubscribe(1)
--> onNext(1): 999
--> onCompleted(1)
--------------------------------------------
----> doOnLifecycle onSubscribe(2)
--> onSubscribe(2)
--> onNext(2): 999
----> doOnLifecycle onDispose(2)
--------------------------------------------
----> doOnNext(3): 999
--> accept(3): 999
--------------------------------------------
--> onSubscribe(4)
----> doOnEach(4) onNext: 999
--> onNext(4): 999
----> doOnEach(4) onComplete
--> onCompleted(4)
--------------------------------------------
--> onNext(5): 999
----> doAfterNext(5): 999
--------------------------------------------
--> onSubscribe(6)
----> doOnError(6): java.lang.Exception: Test Error!
--> onError(6): java.lang.Exception: Test Error!
--------------------------------------------
--> accept(7): 999
----> doOnTerminate(7)
--------------------------------------------
--> onSubscribe(8)
--> onNext(8): 999
----> doOnComplete(8)
--> onComplete(8)
--------------------------------------------
--> onSubscribe(9)
--> onNext(9): 999
--> onComplete(9)
----> doFinally(9)
--------------------------------------------
--> onSubscribe(10)
----> doOnDispose(10)

Javadoc: doOnSubscribe(Consumer onSubscribe)
Javadoc: doOnLifecycle(Consumer onSubscribe, Action onDispose)
Javadoc: doOnNext(Consumer onNext)
Javadoc: doOnEach(Observer observer)
Javadoc: doAfterNext(Consumer onAfterNext)
Javadoc: doOnError(Consumer onError)
Javadoc: doOnTerminate(Action onTerminate)
Javadoc: doAfterTerminate(Action onFinally)
Javadoc: doOnComplete(Action onComplete)
Javadoc: doFinally(Action onFinally)
Javadoc: doOnDispose(Action onDispose)

3. SubscribeOn

指定Observable自身在哪個調度器上執行。

使用調度器 Scheduler 來管理多線程環境中Observable的轉場。你可以使用 SubscribeOn 操作符指定Observable在一個特定的調度器上運轉。

示例代碼：

    // 查看當前線程id
    System.out.println("----> main: threadID = " + Thread.currentThread().getId());

    /**
     *  subscribeOn(Scheduler scheduler)
     *  指定Observable在指定的scheduler上調度
     */
    Observable.create(new ObservableOnSubscribe<Integer>() {
        @Override
        public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
        	// 查看Observable的工作線程id
            System.out.println("----> SubscribeOn: threadID = " + Thread.currentThread().getId());
            emitter.onNext(999);
            emitter.onComplete();
        }
    }).subscribeOn(Schedulers.newThread())  // 指定Observable的工作線程在子線程
            .subscribe(new Consumer<Integer>() {
                @Override
                public void accept(Integer integer) throws Exception {
                    System.out.println("--> accept: " + integer);
                }
            });

    System.in.read();

輸出：

----> main: threadID = 1
----> SubscribeOn: threadID = 13
--> accept: 999

Javadoc: subscribeOn(Scheduler scheduler)

4. ObserverOn

指定一個觀察者在哪個調度器上觀察這個Observable。

使用調度器 Scheduler 來管理多線程環境中Observable的轉場。你可以使用 ObserveOn 操作符指定Observable在一個特定的調度器上發送通知給觀察者 (調用觀察者的onNext, onCompleted , onError 方法)。

示例代碼：

    // 查看當前線程id
    System.out.println("----> main: threadID = " + Thread.currentThread().getId());

    /**
     *  observeOn(Scheduler scheduler,
     *  boolean delayError,     // 可選參數是否延遲異常
     *  int bufferSize          		  // 指定緩存大小
     *  )
     * 指定觀察者在指定的scheduler線程中調度
     */
    Observable.just(999).observeOn(Schedulers.newThread(), true, 3)
            .subscribe(new Consumer<Integer>() {
                @Override
                public void accept(Integer integer) throws Exception {
                	// 查看觀察者的線程id
                    System.out.println("--> accept ThreadID: " + Thread.currentThread().getId());
                    System.out.println("--> accept: " + integer);
                }
            });

    System.in.read();

輸出：

----> main: threadID = 1
--> accept ThreadID: 13
--> accept: 999

Javadoc: observeOn(Scheduler scheduler)
Javadoc: observeOn(Scheduler scheduler, boolean delayError)
Javadoc: observeOn(Scheduler scheduler, boolean delayError, int bufferSize)

5. Serialize

強制一個Observable連續調用並保證行為正確，其實就是同步事件操作。

一個Observable可以異步調用它的觀察者的方法，可能是從不同的線程調用。這可能會讓Observable行為不正確，它可能會在某一個 onNext 調用之前嘗試調用 onCompleted 或 onError 方法，或者從兩個不同的線程同時調用 onNext 方法。使用 Serialize 操作符，你可以糾正這個Observable的行為，保證它的行為是正確的且是同步的。

示例代碼：

    /**
     *  serialize()
     *  強制一個Observable連續調用(同步)並保證行為正確
     */
    Observable.create(new ObservableOnSubscribe<Integer>() {
        @Override
        public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
        	// 多線程事件調用
            new Thread(new Runnable() {
                @Override
                public void run() {
                    for (int i = 0; i < 10; i++) {
                        emitter.onNext(i + 1);
                    }
                    emitter.onComplete();
                }
            }).start();

			// 多線程事件調用
            new Thread(new Runnable() {
                @Override
                public void run() {
                    for (int i = 100; i < 110; i++) {
                        emitter.onNext(i + 1);
                    }
                    emitter.onComplete();
                }
            }).start();
        }
    }).serialize() // 序列化,合法性操作
      .subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                System.out.println("--> onSubscribe");
            }

            @Override
            public void onNext(Integer integer) {
                System.out.println("--> onNext: " + integer);
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("--> onError: " + e);
            }

            @Override
            public void onComplete() {
                System.out.println("--> onComplete");
            }
      });

    System.in.read();

輸出：

---------------------------------------------
下面是沒有使用Serialize()場景，發現不合法的調用
--> onSubscribe
--> onNext: 1
--> onNext: 2
--> onNext: 3
--> onNext: 4
--> onNext: 5
--> onNext: 6
--> onNext: 7
--> onNext: 8
--> onNext: 101
--> onNext: 102
--> onNext: 103
--> onNext: 104
--> onNext: 105
--> onNext: 9
--> onNext: 106
--> onNext: 10
--> onNext: 107
--> onComplete
--> onNext: 108		// 不合法的調用
----------------------------------------------
下面是使用Serialize()場景，合法的事件調用
--> onSubscribe
--> onNext: 1
--> onNext: 2
--> onNext: 3
--> onNext: 4
--> onNext: 5
--> onNext: 6
--> onNext: 7
--> onNext: 8
--> onNext: 9
--> onNext: 101
--> onNext: 102
--> onNext: 103
--> onNext: 104
--> onNext: 105
--> onNext: 106
--> onNext: 107
--> onNext: 108
--> onNext: 109
--> onNext: 110
--> onComplete

Javadoc: serialize()

6. Materialize

Materialize 將數據項和事件通知都當做數據項發射。

一個合法的有限的Obversable將調用它的觀察者的 onNext 方法零次或多次，然后調用觀察者的 onCompleted 或 onError 正好一次。 Materialize 操作符將這一系列調用，包括原來的 onNext 通知和終止通知onCompleted 或 onError 都轉換為一個Observable發射的數據序列。

解析： 將來自原始Observable的通知轉換為 Notification 對象，然后它返回的Observable會發射這些數據。

示例代碼：

    /**
     *  materialize()
     *  將來自原始Observable的通知轉換為Notification對象，然后它返回的Observable會發射這些數據。
     */
    Observable.create(new ObservableOnSubscribe<Integer>() {
        @Override
        public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
            emitter.onNext(1);
            emitter.onNext(2);
            emitter.onError(new Exception("Test Error!"));
            emitter.onComplete();
        }
    }).materialize()
      .subscribe(new Observer<Notification<Integer>>() {
            @Override
            public void onSubscribe(Disposable d) {
                System.out.println("--> onSubscribe");
            }

            @Override
            public void onNext(Notification<Integer> integerNotification) {
                System.out.println("--> onNext: " + integerNotification);
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("--> onError: " + e);
            }

            @Override
            public void onComplete() {
                System.out.println("--> onComplete");
            }
      });

輸出：

--> onSubscribe
--> onNext: OnNextNotification[1]
--> onNext: OnNextNotification[2]
--> onNext: OnErrorNotification[java.lang.Exception: Test Error!]
--> onComplete

Javadoc: materialize()

7. Dematerialize

Dematerialize 操作符是 Materialize 的逆向過程，它將 Materialize 轉換的結果還原成它原本的形式。

解析： dematerialize 反轉這個過程，將原始Observable發射的 Notification 對象還原成Observable的通知。

示例代碼：

    /**
     *  dematerialize()
     *  過時的方法，在Rxjava：2.2.4中已經被dematerialize(Function<T, Notification<R>> selector)替代
     *  將原始Observable發射的 Notification 對象還原成Observable的通知。
     */
    Observable.create(new ObservableOnSubscribe<Integer>() {
        @Override
        public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
            emitter.onNext(1);
            emitter.onNext(2);
            emitter.onError(new Exception("Test Error!"));
            emitter.onComplete();
        }
    }).materialize()
      .dematerialize()  // 將Notification 對象還原成Observable的通知
      .subscribe(new Observer<Object>() {
            @Override
            public void onSubscribe(Disposable d) {
                System.out.println("--> onSubscribe(1)");
            }

            @Override
            public void onNext(Object o) {
                System.out.println("--> onNext(1): " + o);
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("--> onError(1): " + e);
            }

            @Override
            public void onComplete() {
                System.out.println("--> onComplete(1)");
            }
      });

    System.out.println("------------------------------------------------");
    /**
     *  dematerialize(Function<T, Notification<R>> selector)
     *  將原始Observable發射的 Notification 對象經過一個selector函數處理后，發射一個新的Notification，
     *  還原成Observable的通知。
     */
    Observable.create(new ObservableOnSubscribe<Integer>() {
        @Override
        public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
            emitter.onNext(1);
            emitter.onNext(2);
            emitter.onError(new Exception("Test Error!"));
            emitter.onComplete();
        }
    }).materialize()
      .dematerialize(new Function<Notification<Integer>, Notification<Integer>>() {
                @Override
                public Notification<Integer> apply(Notification<Integer> integerNotification) throws Exception {
                    System.out.println("--> apply(2): " + integerNotification);
                    return integerNotification;
                }
      }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                System.out.println("--> onSubscribe(2)");
            }

            @Override
            public void onNext(Integer integer) {
                System.out.println("--> onNext(2): " + integer);
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("--> onError(2): " + e);
            }

            @Override
            public void onComplete() {
                System.out.println("--> onComplete(2)");
            }
    });

輸出：

--> onSubscribe(1)
--> onNext(1): 1
--> onNext(1): 2
--> onError(1): java.lang.Exception: Test Error!
------------------------------------------------
--> onSubscribe(2)
--> apply(2): OnNextNotification[1]
--> onNext(2): 1
--> apply(2): OnNextNotification[2]
--> onNext(2): 2
--> apply(2): OnErrorNotification[java.lang.Exception: Test Error!]
--> onError(2): java.lang.Exception: Test Error!

Javadoc: dematerialize()
Javadoc: dematerialize(Function<T,Notification<R>> selector)

接續：

后續的Observable的輔助操作部分請參考： Rxjava2 Observable的輔助操作詳解及實例（二）

Rx介紹與講解及完整目錄參考：Rxjava2 介紹與詳解實例