RxJava操作符(十)自定义操作符
发布于 2016-02-05    1,019 次阅读
Rxjava允许我们来自定义操作符来满足我们特殊的需求。如果我们的自定义操作符想要作用到Observable发射出来的数据上,我们就要使用lift操作符;如果我们的自定义操作符想要改变整个的Observable,就需要使用compose操作符了。 一、 lift 当我们自定义了一个操作符后,使用lift可以将我们自定义的操作符和其它的操作符一起做链式调用,就好像Rxjava原生的操作符一样。下面我...

Rxjava允许我们来自定义操作符来满足我们特殊的需求。如果我们的自定义操作符想要作用到Observable发射出来的数据上,我们就要使用lift操作符;如果我们的自定义操作符想要改变整个的Observable,就需要使用compose操作符了。

一、 lift

当我们自定义了一个操作符后,使用lift可以将我们自定义的操作符和其它的操作符一起做链式调用,就好像Rxjava原生的操作符一样。下面我们自定义一个操作符,并使用lift添加到Observable的链式调用里面:

private Observable<String> liftObserver() {
    Operator<String, String> myOperator = new Operator<String, String>() {
        @Override
        public Subscriber<? super String> call(Subscriber<? super String> subscriber) {
            return new Subscriber<String>(subscriber) {
                @Override
                public void onCompleted() {
                    if (!subscriber.isUnsubscribed()) {
                        subscriber.onCompleted();
                    }
                }

                @Override
                public void onError(Throwable e) {
                    if (!subscriber.isUnsubscribed()) {
                        subscriber.onError(e);
                    }
                }

                @Override
                public void onNext(String s) {
                    if (!subscriber.isUnsubscribed()) {
                        subscriber.onNext("myOperator:" + s);
                    }
                }
            };
        }

    };
    return Observable.just(1, 2, 3).map(integer -> "map1:" + integer).lift(myOperator).map(s -> "map2:" + s);
}

我们使用lift将我们自定义的操作符夹在两个map操作符之间,形成链式的调用。
对其进行订阅:

mLButton.setText("lift");
mLButton.setOnClickListener(e -> liftObserver().subscribe(s -> log("lift:" + s)));

运行结果如下,可以看到输出的结果也正如预期的一样,我们自定义的操作符作用在两个map操作符之间。

二、compose

Compose操作符是将源Observable按照自定义的方式转化成另外一个新的Observable。可以这么说compose是对Observable进行操作的而lift是对Subscriber进行操作的,作用点是不同的。

下面我们自定义一个Transformer对象,并使用compse操作符将其应用到Observable上:

private Observable<String> composeObserver() {
    Transformer<Integer, String> myTransformer = new Transformer<Integer, String>() {
        @Override
        public Observable<String> call(Observable<Integer> integerObservable) {
            return integerObservable
                    .map(integer -> "myTransforer:" + integer)
                    .doOnNext(s -> log("doOnNext:" + s));
        }
    };
    return Observable.just(1, 2, 3).compose(myTransformer);
}

在这个我们自定义的Transformer对象中,我们将原先发射Integer的Observable转化成了一个发射String内容的Observable;并且我们还添加了doOnNext操作符来打印出发射的数据,方便我们进行后续的调试。个人感觉Transformer更像是一个批量转化器,如你有很多Observable对象在使用,可以定义一个通用的Transformer对象,里面可以通过doOnNext打印数据,可以定义subscribeOn和observeOn的线程等等,最后使用compose操作符将其应用到我们所有的Observable对象上就可以统一进行设定了。
下面我们进行一下订阅吧:

mRButton.setText("compose");
mRButton.setOnClickListener(e -> composeObserver().subscribe(s -> log("compose:" + s)));

运行的结果如下,我们可以看到对每个数据都将其转化为String之后通过doOnNext输出了数据的值,然后输出了Subscriber接收到的数据。

关于自定义操作符还有一些需要注意的地方:

  • 自定义Operator在发射任何数据之前都要使用!subscriber.isUnsubscribed()来检查Subscriber的状态,如果没有任何Subscriber订阅就没有必要去发射数据了
  • 自定义Operator要遵循Observable的核心原则:

    • 可以多次调用Subscriber的onNext方法,但是同一个数据只能调用一次。
    • 可以调用Subscriber的onComplete或者onError方法,但是这两个是互斥的,调用了一个就不能再调用另外一个了,并且一旦调用了任何一个方法就不能再调用onNext方法了。
    • 如果无法保证遵守以上两条,可以对自定义操作符加上serialize操作符,这个操作符会强制发射正确的数据。
  • 自定义Operator内部不能阻塞住。

  • 如果通过compose组合多个操作符就能达到目的就不要自己去写新的代码来实现,在Rxjava的源码中就有很多这样的例子,如:
    • first()操作符是通take(1).single()来实现的。
    • ignoreElements()是通过 filter(alwaysFalse())来实现的。
    • reduce(a) 是通过 scan(a).last()来实现的。
  • 当有异常的时候,不能继续发射正常的数据,要立刻调用Subscriber的onError方法将异常抛出去。
  • 注意发射数据为null的情况,这和完全不发射数据不是一回事。

关于自定义操作符就到这里了,本文的demo程序见github

版权说明 : 本文为转载文章, 版权为原作者所有

原文标题 : RxJava操作符(十)自定义操作符

原文连接 : http://mushuichuan.com/2016/02/05/rxjava-operator-10/