2. 程式人生 > >7章 RxJava高階用法（一）

7章 RxJava高階用法（一）

阿新 • • 發佈：2018-12-16

CSDN學院課程地址

7. RxJava高階用法（一）

7.1 自定義Operator

自定義Operator屬於RxJava的高階用法，可以自己自定義一些適用於常見應用場景的操作符。實現自定義Operator很簡單，只需要實現RxJava提供的ObservableOperator

介面，實現對應的功能即可，同時，使用lift操作符將自定義操作符應用到我們的程式中。下面我們使用自定義Operator，該操作符的作用是將List集合轉換成String型別的輸出

1、實現ObservableOperator，建立自定義Operator

public class CustomOperator implements ObservableOperator<String, List<String>> {

    @Override
    public Observer<? super List<String>> apply(final Observer<? super String> observer) throws Exception {
        return new Observer<List<String>>() {
            @Override
            public void onSubscribe(Disposable d) {
                observer.onSubscribe(d);
            }

            @Override
            public void onNext(List<String> strings) {
                observer.onNext(strings.toString());
            }

            @Override
            public void onError(Throwable e) {
                observer.onError(e);
            }

            @Override
            public void onComplete() {
                observer.onComplete();
            }
        };
    }
}

2、使用lift操作符新增自定義Operator

public class Main {
    public static void main(String[] args) {
        //建立被觀察者
        Observable.create(new ObservableOnSubscribe<List<String>>() {
            @Override
            //預設在主執行緒裡執行該方法
            public void subscribe(@NonNull ObservableEmitter<List<String>> e) throws Exception {
                ArrayList<String> list = new ArrayList<>();
                list.add("1");
                list.add("2");
                list.add("3");
                list.add("4");
                e.onNext(list);
                e.onComplete();
            }
        })
                .lift(new CustomOperator())
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(String s) {
                        System.out.println("onNext=" + s);
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

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

3、輸出結果

onNext=[1, 2, 3, 4]
onComplete

7.2 自定義Transformer

自定義Transformer表示一個批量操作符的變換器，如果你在很多Observable中使用相同的一系列操作符，比如每次都要使用到map+take+doOnNext等操作，那麼就可以定義一個通用的Transformer物件，裡面可以將需要重複用到的操作符打包成Transformer物件，使用compose操作符將Transformer物件應用到我們的Observable上即可

實現自定義Transformer很簡單，只需要實現RxJava提供的ObservableTransformer介面，實現對應的功能即可，同時，使用compose操作符將自定義Transformer應用到我們的程式中。下面我們使用自定義Transformer，該Transformer的作用是將發射的資料從Integer轉換成String，並取2個數據項，同時在發射的時候監聽發射事件，進行輸出的列印

1、實現ObservableTransformer，建立自定義Transformer

public class CustomTransformer implements ObservableTransformer<Integer, String> {
    @Override
    public ObservableSource<String> apply(io.reactivex.Observable<Integer> upstream) {
        return upstream.take(2).map(new Function<Integer, String>() {
            @Override
            public String apply(Integer integer) throws Exception {
                return "序號：" + integer + "發射成功";
            }
        }).doOnNext(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                System.out.println(s + "，準備發射");
            }
        });
    }
}

2、使用compose操作符新增自定義Transformer

public class Main {

    public static void main(String[] args) {
        //建立被觀察者
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
            }
        })
                .compose(new CustomTransformer())
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(String s) {
                        System.out.println(s);
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });
    }
}

3、輸出結果

序號：1發射成功，準備發射
序號：1發射成功
序號：2發射成功，準備發射
序號：2發射成功

在安卓開發中，通常我們也會自定義Transformer來實現我們常用的執行緒切場景，具體如下

public static <T> ObservableTransformer<T, T> schedulersTransformer() {
    return new ObservableTransformer<T, T>() {
        @Override
        public ObservableSource<T> apply(@NonNull Observable<T> upstream) {
            return upstream.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread());
        }
    };
}

public static <T> FlowableTransformer<T, T> schedulersTransformerForFlowable() {
    return new FlowableTransformer<T, T>() {
        @Override
        public Publisher<T> apply(Flowable<T> upstream) {
            return upstream.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread());
        }
    };
}

7.3 自定義Plugin

自定義Plugin表示自定義外掛，自定義外掛可以在RxJavaPlugins中提供的介面中去插入自己的一段程式碼操作，類似於面向切面程式設計，或者理解成Android的Hook。如果你需要在所有的訂閱事件中去插入一段統一的操作，或者是監聽所有訂閱事件發生異常時的回撥，都可以使用自定義外掛。在實際應用中，目前並未發現有什麼作用

實現自定義Plugin只需要呼叫RxJavaPlugins提供的set方法即可，下面我們通過例子輸出Observable和Observer的地址資訊，來驗證每次訂閱的時候，回撥自定義Plugin的方法中，外掛物件和源物件是否為同一個物件

1、通過設定ObservableSubscribe，每次對Observable操作的時候回撥

public class Main {

    public static void main(String[] args) {

        RxJavaPlugins.setOnObservableAssembly(new CustomObservableAssembly());//任意操作符都有回撥
        RxJavaPlugins.setOnObservableSubscribe(new CustomObservableSubscribe());//每次subscribe時候有回撥

        Observable observable = getObservable();
        Observer<Integer> observer = getObserver();

        System.out.println("main observable.toString:" + observable.toString());
        System.out.println("main observer.toString:" + observer.toString());

        observable.subscribe(observer);
    }

    public static Observable getObservable() {
        return Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(5);
                emitter.onNext(2);
                emitter.onNext(3);
            }
        });
    }

    public static Observer<Integer> getObserver() {
        return new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

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

            @Override
            public void onError(Throwable e) {
                System.out.println(e.getMessage());
            }

            @Override
            public void onComplete() {

            }
        };
    }
}

2、CustomObservableAssembly

public class CustomObservableAssembly implements Function<Observable, Observable> {
    @Override
    public Observable apply(Observable observable) throws Exception {
        System.out.println("CustomObservableAssembly observable.toString:" + observable.toString());
        observable.take(2);
        return observable;
    }
}

3、CustomObservableSubscribe

public class CustomObservableSubscribe implements BiFunction<Observable, Observer, Observer> {
    @Override
    public Observer apply(Observable observable, Observer observer) throws Exception {
        System.out.println("CustomObservableSubscribe observable.toString:" + observable.toString() + ",observer.toString:" + observer.toString());
        return observer;
    }
}

4、輸出結果

地址相同說明是同個物件，自定義外掛Hook成功

CustomObservableAssembly observable.toString:[email protected]a7ca
main observable.toString:[email protected]a7ca
main observer.toString:[email protected]
CustomObservableSubscribe observable.toString:[email protected]a7ca,observer.toString:[email protected]
onNext=5
onNext=2
onNext=3

補充：

可以通過設定ErrorHandler，發生異常時會回撥

RxJavaPlugins.setErrorHandler();

可以通過設定SchedulerHandler來Hook到對應的schedule

RxJavaPlugins.setIoSchedulerHandler();
RxJavaPlugins.setNewThreadSchedulerHandler();
RxJavaPlugins.setComputationSchedulerHandler();
RxJavaPlugins.setSingleSchedulerHandler();

錯誤演示：

由於CustomObservableAssembly是在任意操作符操作的時候都會回撥，所以在回撥裡面是不可以對observable再進行操作符的操作，否則回撥裡面observable的操作符還是會回撥CustomObservableAssembly自身，導致死迴圈，發生StackOverflowError

public class CustomObservableAssembly implements Function<Observable, Observable> {
    @Override
    public Observable apply(Observable observable) throws Exception {
        System.out.println("CustomObservableAssembly observable.toString:" + observable.toString());
        observable.take(2);
        return observable;
    }
}

由於CustomObservableSubscribe是在subscribe之後進行的回撥，如果在回撥裡面對observable進行操作符的操作，這個時候是不會生效的，因為在subscribe之後onNext的函式是不會再處理後面新添的操作符，原理與原始碼有關

public class CustomObservableSubscribe implements BiFunction<Observable, Observer, Observer> {
    @Override
    public Observer apply(Observable observable, Observer observer) throws Exception {
        System.out.println("CustomObservableSubscribe observable.toString:" + observable.toString() + ",observer.toString:" + observer.toString());
        observable.take(2);
        return observer;
    }
}

相關推薦

7章 RxJava高階用法（一）

CSDN學院課程地址 RxJava2從入門到精通-初級篇:https://edu.csdn.net/course/detail/10036 RxJava2從入門到精通-中級篇:https://edu.csdn.net/course/detail/10037

8章 RxJava高階用法（二）

CSDN學院課程地址 RxJava2從入門到精通-初級篇:https://edu.csdn.net/course/detail/10036 RxJava2從入門到精通-中級篇:https://edu.csdn.net/course/detail/10037

很多人不瞭解的嵌入式C高階用法（一）

本文來自網路 1、記憶體管理 我們需要知道——變數，其實是記憶體地址的一個抽像名字罷了。在靜態編譯的程式中，所有的變數名都會在編譯時被轉成記憶體地址。機器是不知道我們取的名字的，只知道地址。  記憶體的使用時程式設計中需要考慮的重要因素之一，這不僅由於系統記憶體是有限

【深入Java基礎】HashMap高階用法（一）：排序

HashMap高階用法（一）：排序 根據key排序 HashMap是無序的，我們可以根據key進行升序或降序。 1.利用List和Collections來實現排序 先獲取HashMap的keySet，然後將keySet放入List，在由Collectio

block的高階用法（一） -- block傳值

我們先從簡單的入手，先講下block的傳值，後面會有更精彩的。 在講傳值之前我們先複習下block的基礎知識： - (void)viewDidLoad { [super viewDidLoa

階段1-第三章-webpack-高階配置（一）

1.HTML 中img標籤的圖片資源處理 html-withimg-loader 只要在html中正常引用圖片即可，webpac

算法導論筆記——第十~十一章 數據結構（一） 散列

發生 情況 要求 sub 裝載 ted 因子 let 完全 第十章 基本數據結構 棧：可由數組表示 隊列：可由數組表示 指針和對象：可由多數組表示。可用棧表示free list 有根數： 　　二叉樹：左右孩子 　　分支無限制：左孩子右兄弟表示法 第十一章 散列表 數組：

Linux基礎之常見命令用法（一）

linux基礎命令入門(一)一、Linux文件目錄結構 在講述之前，先簡短的說說Windows文件結構，打開‘計算機’，看到的一個個的驅動器(盤符，例C盤、D盤等)，點開其中任意盤符，看到的是一個個文件或文件夾，繼續打開...,每個盤都有自己的根目錄。若是把其打開過程畫下來，便可得到如下多棵倒樹並列的圖

Redis Sentinel機制與用法（一）

clear monitor 拒絕 coff failure abstract trac 用法 XML from:https://my.oschina.net/dyyweb/blog/513680 摘要: Redis Sentinel機制與用法（一） Sentin

golang 中io包用法（一）

不同的 下一個 同時 顯示 cte () ews 執行 忽略 本文轉自Golove博客：http://www.cnblogs.com/golove/p/3276678.html io 包為I/O原語提供了基礎的接口.它主要包裝了這些原語的已有實現，如 os 包中的

tensorboard 用法（一）

import tensorflow as tf import numpy as np ## prepare the original data with tf.name_scope('data'): x_data = np.random.rand(100).astype(np

ActiveMQ學習筆記（15）----Message Dispatch高階特性（一）

1. Message Cursors 　　1.1 概述 　　ActiveMQ傳送持久化訊息的典型的厝裡方式是：當訊息的消費者準備就緒時，訊息傳送系統把儲存的訊息按批次傳送給消費者，在傳送完一個批次的訊息後，指標的標記位置指向下一個批次的待發訊息的位置，進行後續的傳送操作。這是一種 比較健壯和靈活的訊息傳送

ActiveMQ學習筆記（19）----Consumer高階特性（一）

1. Exclusive Consumer 　　獨有消費者：Queue中的訊息是按照順序被分發到consumer的，然而，當你有多個consumers同時從相同的queue中提取訊息時，你將失去這個保證。因為這些訊息是被多個執行緒併發的處理。有的時候，保證訊息按照順序處理是很重要的。例如：你可能不希望在插入

第2章 RFID基礎知識（一）

1.RFID即射頻識別。（常稱為電子標籤）RFID射頻識別是一種非接觸式的自動識別技術，識別高速運動物體並可同時識別多個標籤，識別距離可達幾十米。 2.RFID的組成：一套完整的RFID系統必須由標籤、閱讀器和天線組成。 3.電子標籤的分類：標籤類、注塑類、卡片類。 4.閱讀器與電子標籤之間的射頻訊

python高階語法（一）

文章目錄 GIL全域性直譯器鎖 什麼是GIL全域性直譯器鎖 執行緒何時會釋放GIL鎖 如何避免這種現象 多執行緒單執行緒的選擇 深拷貝~淺拷貝 按資料型別分類 1、 拷貝單一的可變

第7章 網路層協議（3）_ARP協議

3. ARP協議 3.1 ARP（Address Resolution Protocol）協議的工作過程和安全隱患 （1）計算機A和C通訊之前，先檢查ARP快取中是否有計算機C的IP地址對應的MAC地址。如果沒有，就啟用ARP協議傳送一個ARP廣播請求解析192.168.0.4的MAC

第7章 網路層協議（2）_ICMP協議

2. ICMP協議 2.1 ICMP報文（Internet Control Message Protocol）的型別 報文型別 型別值 程式碼 描述 請求報文 8 0

第7章 網路層協議（1）_網路層首部

1. 網路層首部 1.1 網路層協議 （1）TCP/IP協議棧網路層的4個協議：IP協議、ICMP協議、IGMP協議和ARP協議。 （2）IP協議：動態路上協議的統稱，包括RIP和OSPF協議。 （3）TCP/IP協議分成四層：應用層定義了客戶端和伺服器通訊規範，傳輸層實現可靠

Python高階教程（一）面向物件

Python 面向物件 Python從設計之初就已經是一門面向物件的語言，正因為如此，在Python中建立一個類和物件是很容易的。本章節我們將詳細介紹Python的面向物件程式設計。 如果你以前沒有接觸過面向物件的程式語言，那你可能需要先了解一些面嚮物件語言的一些基本特徵，在頭腦裡頭形

django之ORM介紹與基本用法（一）

  一、ORM介紹 1.什麼是ORM ORM 全拼Object-Relation Mapping. 中文意為 物件-關係對映. 在MVC/MVT設計模式中的Model模組中都包括ORM 2.ORM優勢 （1）只需要面