Rxjava2入门教程八：自定义操作符

如需下载源码，请访问
https://github.com/fengchuanfang/Rxjava2Tutorial
本篇文章对应的Demo类为：com.edward.edward.Rxjava2Demo.Rxjava2_8_Custom_Operators
文章原创，转载请注明出处：
RxJava入门教程八：自定义操作符
运行源码，打开app，点击Demo图标，可进入以下页面

Rxjava2Tutorial8.png

在RxJava中，我们可以通过自定义操作符来拓展Observable内置操作符无法实现的功能，用以满足我们特殊的需求。
我们不需要继承Observable或者实现ObservableSource通过增加方法来拓展可观察对象类的功能，根据拓展功能的不同我们可以通过以下两种方式：

序列操作符（Sequence Operators）的拓展：如果我们的操作符是为了对Observable发射的数据起作用，我们需要实现ObservableOperator接口，并通过lift()操作符与Observable内置操作符做链式调用。

转换操作符（Transformational Operators）的拓展：如果我们的操作符是为了对整个Observable（将Observable作为一个整体）起作用，我们需要实现ObservableTransformer接口，并通过compose()操作符与Observable内置操作符做链式调用。

序列操作符（Sequence Operators）

下面我们实现一个平方计算的操作符来看一下如何自定义一个序列操作符

首先，实现ObservableOperator接口，代码如下：

public class SQOperator implements ObservableOperator<Integer, Integer> {

        @Override
        public Observer<? super Integer> apply(Observer<? super Integer> observer) throws Exception {
            return new Observer<Integer>() {

                private Disposable mDisposable;

                @Override
                public void onSubscribe(Disposable d) {
                    mDisposable = d;
                    observer.onSubscribe(d);
                }

                @Override
                public void onNext(Integer integer) {
                    if (!mDisposable.isDisposed()) observer.onNext(integer * integer);
                }

                @Override
                public void onError(Throwable e) {
                    if (!mDisposable.isDisposed()) observer.onError(e);
                }

                @Override
                public void onComplete() {
                    if (!mDisposable.isDisposed()) observer.onComplete();
                }
            };
        }
    }

然后，通过lift()操作符，与Observable内置操作符做链式调用

public void demo1() {
        Observable
                .just(1, 2, 3, 4)
                .lift(new SQOperator())
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(Integer integer) {
                        SystemOut.println(integer);
                    }

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

                    @Override
                    public void onComplete() {
                        SystemOut.println("完成");
                    }
                });
    }

运行结果如下：

执行结果为-->
1
4
9
16
完成

在上面的例子中，自定义操作符并不改变发射数据的个数，下面我们再实现一个拆分字符串的操作符，在此操作符中会改变发射数据的个数。

public class ToCharOperator implements ObservableOperator<Character, String> {


        @Override
        public Observer<? super String> apply(Observer<? super Character> observer) throws Exception {
            return new Observer<String>() {
                private Disposable mDisposable;

                @Override
                public void onSubscribe(Disposable d) {
                    mDisposable = d;
                }

                @Override
                public void onNext(String s) {
                    for (Character character : s.toCharArray()) {
                        if (mDisposable.isDisposed()) break;
                        observer.onNext(character);//下游Observer多次调用onNext方法
                    }
                }

                @Override
                public void onError(Throwable e) {
                    if (!mDisposable.isDisposed()) observer.onError(e);
                }

                @Override
                public void onComplete() {
                    if (!mDisposable.isDisposed()) observer.onComplete();
                }
            };
        }
    }

public void demo2() {
        Observable.just("Hello", "Rxjava")
                .lift(new ToCharOperator())
                .map(new Function<Character, String>() {
                    @Override
                    public String apply(Character character) throws Exception {
                        SystemOut.println(character);
                        return character.toString();
                    }
                })
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

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

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

                    @Override
                    public void onComplete() {
                        SystemOut.println("完成");
                    }
                });

    }

运行后结果如下：

执行结果为-->
72
H
101
e
108
l
108
l
111
o
82
R
120
x
106
j
97
a
118
v
97
a
完成

转换操作符（Transformational Operators）

下面我们同样通过实现一个平方计算的例子来看一下如何自定义一个转换操作符
首先，实现ObservableTransformer接口，代码如下：

public class SQTransFormer implements ObservableTransformer<Integer, Integer> {


        @Override
        public ObservableSource<Integer> apply(Observable<Integer> upstream) {
            return upstream.map(new Function<Integer, Integer>() {
                @Override
                public Integer apply(Integer integer) throws Exception {
                    return integer * integer;
                }
            });
        }
    }

由于在自定义转换操作符中，Observable是作为一个整体出现的，所以我们只能通过调用Observable现有的方法，对发射的数据做出改变。
然后通过compose()操作符，与Observable内置操作符做链式调用

public void demo3() {
        Observable.just(1, 2, 3, 4)
                .compose(new SQTransFormer())
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(Integer integer) {
                        SystemOut.println(integer);
                    }

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

                    @Override
                    public void onComplete() {
                        SystemOut.println("完成");
                    }
                });
    }

运行结果如下：

执行结果为-->
1
4
9
16
完成

通过自定义转换操作符可以对Observable的一些操作符的固定的链式调用进行封装，例如
在Android开发中，线程转换通常具有固定的形式，如下：

public void demo4() {
        Observable
                .fromCallable(new Callable<String>() {
                    @Override
                    public String call() throws Exception {
                        SystemOut.println("call:当前线程-->" + Thread.currentThread().getName());
                        return "Hello Rxjava";
                    }
                })
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .onTerminateDetach()//订阅关系取消时，解除上游对下游的引用
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        SystemOut.println("onSubscribe:当前线程-->" + Thread.currentThread().getName());
                    }

                    @Override
                    public void onNext(String s) {
                        SystemOut.println("onNext:当前线程-->" + Thread.currentThread().getName());
                        SystemOut.println(s);
                    }

                    @Override
                    public void onError(Throwable e) {
                        SystemOut.println("onError:当前线程-->" + Thread.currentThread().getName());
                    }

                    @Override
                    public void onComplete() {
                        SystemOut.println("onComplete:当前线程-->" + Thread.currentThread().getName());
                    }
                });
    }

运行结果如下：

执行结果为-->

onSubscribe:当前线程-->main
call:当前线程-->RxCachedThreadScheduler-2
onNext:当前线程-->main
Hello Rxjava
onComplete:当前线程-->main

通常会在异步线程中执行耗时操作，然后再UI线程中进行数据显示,对于固定的链式组合

    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .onTerminateDetach()//订阅关系取消时，解除上游对下游的引用

我们可以通过一个自定义转换操作符，进行封装，代码如下：

public class SchedulersTransFormer<T> implements ObservableTransformer<T, T> {

        @Override
        public ObservableSource<T> apply(Observable<T> upstream) {
            return upstream
                    .subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .onTerminateDetach();//订阅关系取消时，解除上游对下游的引用;
        }
    }

通过自定义转换操作符SchedulersTransFormer，可将demo4代码简化为


public void demo5() {
        Observable
                .fromCallable(new Callable<String>() {
                    @Override
                    public String call() throws Exception {
                        SystemOut.println("call:当前线程-->" + Thread.currentThread().getName());
                        return "Hello Rxjava";
                    }
                })
                .compose(new SchedulersTransFormer<>())
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        SystemOut.println("onSubscribe:当前线程-->" + Thread.currentThread().getName());
                    }

                    @Override
                    public void onNext(String s) {
                        SystemOut.println("onNext:当前线程-->" + Thread.currentThread().getName());
                        SystemOut.println(s);
                    }

                    @Override
                    public void onError(Throwable e) {
                        SystemOut.println("onError:当前线程-->" + Thread.currentThread().getName());
                    }

                    @Override
                    public void onComplete() {
                        SystemOut.println("onComplete:当前线程-->" + Thread.currentThread().getName());
                    }
                });
    }

运行后将会看到与demo4相同的结果：

执行结果为-->

onSubscribe:当前线程-->main
call:当前线程-->RxCachedThreadScheduler-2
onNext:当前线程-->main
Hello Rxjava
onComplete:当前线程-->main

关于自定义操作符官方建议的一些规范

对于自定义序列操作符，在通过onNext()方法向下游发射数据以及通过onComplete()方法发射完成信息或通过onError方法发射异常信息时，需要检查当前订阅关系是否有效，如果当前订阅关系已取消，则没必要再发射数据或信息。例如前面的SQOperator和ToCharOperator操作符中的判断

if (!mDisposable.isDisposed()){
    ...
}

自定义序列操作符，需要遵循可观察对象Observable协议的核心原则：

在一次订阅中，可以调用多次onNext()方法，但是同一数据只能调用一次
在一次订阅中，要保证onCompleted()方法和onError()方法的互斥性，调用了一个，就不能调用另一个，并且一旦调用了其中任何一个，就不能再调用onNext()方法。

如果不能确保自定义操作符完全符合上述核心原则，可以增加serialize()操作符，它可以对不符合规范的序列做强制转换。
我们也可以通过github上的一个单元测试框架Global Contract Tests via Unit Tests #1962 来测试我们的自定义操作符是否符合可观察对象协议

自定义操作符内部不能做堵塞操作

如果通过实现自定义转换操作符ObservableTransformer组合Observable现有的操作符就可以达到目的，就不要自己重新写代码去实现新的操作符。

如果在自定义操作符中，将一个方法或λ表达式作为一个参数传入，需要注意捕获方法或λ表达式中的异常并通过onError方法传递给下游。

对于一些灾难性的异常如果直接调用onError传递给下游的话可能毫无意义或者会使问题变得更严重，我们可以通过 io.reactivex.Exceptions.throwIfFatal(throwable) 方法进行过滤，然后再重新抛出它们。

通常情况下，如果遇到异常就应通过onError方法立马通知，而不是等发射更多的数据后再做处理。

注意null值的处理，因为在RxJava2中不允许将null作为数据发射，在向下游转发数据时，需要将null值过滤掉或者转换成异常通过onError方法进行通知。

新的响应式关系下，自定义操作符的实现

对于其他可观察对象自定义操作符的实现，需要实现各自的对应的序列操作符或转换操作符的接口，以及遵循各自的可观察对象的协议。

Flowable
FlowableOperator<Downstream, Upstream>
FlowableTransformer<Upstream, Downstream>

在自定义Flowable序列操作符时，需要注意对背压操作的处理

Single
SingleOperator<Downstream, Upstream>
SingleTransformer<Upstream, Downstream>

Completable
CompletableOperator
CompletableTransformer

Maybe
MaybeOperator<Downstream, Upstream>
MaybeTransformer<Upstream, Downstream>

下面实现一个Single下的平方操作符，代码如下：

public class SQSingleOperator implements SingleOperator<Integer, Integer> {

        @Override
        public SingleObserver<? super Integer> apply(SingleObserver<? super Integer> observer) throws Exception {
            return new SingleObserver<Integer>() {
                private Disposable mDisposable;

                @Override
                public void onSubscribe(Disposable d) {
                    mDisposable = d;
                    observer.onSubscribe(d);
                }

                @Override
                public void onSuccess(Integer integer) {
                    observer.onSuccess(integer * integer);
                }

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

注意：不同于Observable，在调用下游的onSuccess()或onError方法时，不需要对当前订阅关系是否取消进行判断，否则将会收不到结果，这是由Single独特的响应式编程规范决定的，一旦订阅就会有且只有一个结果，要么是onSuccess,要么是onError。
运行以下代码：


public void demo6() {
        Single.just(3)
                .lift(new SQSingleOperator())
                .subscribe(new SingleObserver<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onSuccess(Integer integer) {
                        SystemOut.println(integer);
                    }

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

执行结果为-->

开始订阅
9

最后编辑于：2019.01.02 12:44:24

人面猴
序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
沈念sama阅读 194,457评论 5赞 459
死咒
序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
沈念sama阅读 81,837评论 2赞 371
救了他两次的神仙让他今天三更去死
文/潘晓璐我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
开封第一讲书人阅读 141,696评论 0赞 319
道士缉凶录：失踪的卖姜人
文/不坏的土叔我叫张陵，是天一观的道长。经常有香客问我，道长，这世上最难降的妖魔是什么？我笑而不...
开封第一讲书人阅读 52,183评论 1赞 263
港岛之恋（遗憾婚礼）
正文为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
茶点故事阅读 61,057评论 4赞 355
恶毒庶女顶嫁案：这布局不是一般人想出来的
文/花漫我一把揭开白布。她就那样静静地躺着，像睡着了一般。火红的嫁衣衬着肌肤如雪。梳的纹丝不乱的头发上，一...
开封第一讲书人阅读 46,105评论 1赞 272
城市分裂传说
那天，我揣着相机与录音，去河边找鬼。笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。我是一名探鬼主播，决...
沈念sama阅读 36,520评论 3赞 381
双鸳鸯连环套：你想象不到人心有多黑
文/苍兰香墨我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
开封第一讲书人阅读 35,211评论 0赞 253
万荣杀人案实录
序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
沈念sama阅读 39,482评论 1赞 290
护林员之死
正文独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋以下内容为张勋视角年9月15日...
茶点故事阅读 34,574评论 2赞 309
白月光启示录
正文我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
茶点故事阅读 36,353评论 1赞 326
活死人
序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
沈念sama阅读 32,213评论 3赞 312
日本核电站爆炸内幕
正文年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
茶点故事阅读 37,576评论 3赞 298
男人毒药：我在死后第九天来索命
文/蒙蒙一、第九天我趴在偏房一处隐蔽的房顶上张望。院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
开封第一讲书人阅读 28,897评论 0赞 17
一桩弑父案，背后竟有这般阴谋
文/苍兰香墨我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。一阵脚步声响...
开封第一讲书人阅读 30,174评论 1赞 250
情欲美人皮
我被黑心中介骗来泰国打工，没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。一个月前我还...
沈念sama阅读 41,489评论 2赞 341
代替公主和亲
正文我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
茶点故事阅读 40,683评论 2赞 335

Rxjava2入门教程八：自定义操作符

序列操作符（Sequence Operators）

转换操作符（Transformational Operators）

关于自定义操作符官方建议的一些规范

新的响应式关系下，自定义操作符的实现

推荐阅读更多精彩内容