RxJava（十一）defer 操作符实现代码支持链式调用_rxjava defer

RxJava 系列文章目录导读：

一、RxJava create 操作符的用法和源码分析
二、RxJava map 操作符用法详解
三、RxJava flatMap 操作符用法详解
四、RxJava concatMap 操作符用法详解
五、RxJava onErrorResumeNext 操作符实现 app 与服务器间 token 机制
六、RxJava retryWhen 操作符实现错误重试机制
七、RxJava 使用 debounce 操作符优化 app 搜索功能
八、RxJava concat 操作处理多数据源
九、RxJava zip 操作符在 Android 中的实际使用场景
十、RxJava switchIfEmpty 操作符实现 Android 检查本地缓存逻辑判断
十一、RxJava defer 操作符实现代码支持链式调用
十二、combineLatest 操作符的高级使用
十三、RxJava 导致 Fragment Activity 内存泄漏问题
十四、interval、takeWhile 操作符实现获取验证码功能
十五、RxJava 线程的自由切换

---

一、前言

现在越来越多 Android 开发者使用到 RxJava，在 Android 使用 RxJava 主要有如下好处：

• 轻松切换线程。

• 轻松解决回调的嵌套问题。现在的 App 业务逻辑越来越复杂，多的时候 3，4 层回调嵌套，使得代码可维护性变得很差。RxJava 链式调用使得这些调用变得扁平化。

随着 RxJava 的流行，越来越多的开源项目开始支持 RxJava，像 Retrofit、GreenDao 等。这些开源项目支持 RxJava 使得我们解决复杂业务变得非常方便。

但是这些还不够，有的时候我们自己的封装的业务也需要支持 RxJava，举个例子：查询数据、处理本地文件等操作，总而言之就是一些耗时任务。而且还要处理这些操作的成功、失败、线程切换等操作。如果还是像以前那样做，那就太不优雅了。

二、下面就来探讨下如何使得代码支持 RxJava 风格

遇到这种问题，在我脑海里浮现的第一种方式就是通过 Observable 的 create 操作符。因为在里面我们可以控制数据的发射。就像上一篇文章那样《RxJava switchIfEmpty 操作符实现 Android 检查本地缓存逻辑判断》

如下代码片段：

Observable.create(new Observable.OnSubscribe<Object>() {
            @Override
            public void call(Subscriber<? super Object> subscriber) {
                try {
                    List<Article> as = articleDao.queryBuilder()
                            .where(ArticleDao.Properties.CategoryId.eq(categoryId))
                            .orderDesc(ArticleDao.Properties.Id)
                            .offset((pageIndex - 1) * pageSize)
                            .limit(pageSize).list();
                    if (as == null || as.isEmpty()) {
                        subscriber.onNext(null);
                    }else{
                        subscriber.onNext(as);
                    }
                }catch (Exception e){
                    subscriber.onError(e);
                }
                subscriber.onCompleted();
            }
        });
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

这样确实没有没有问题。但是我们要封装下， 每个方法都这样写维护性和扩展比较差(例如有天我想换种方式来实现而不是 create，如果通过方法封装一下，修改就变得容易多了)
如何封装呢？通过分析知道，大部分代码是相同的，只是我们的业务不一样。那么通过模板方法解决吧。业务方法通过接口回调的方式传递进来，因为我们不知道调用者是什么业务。

回调接口如下（T表示我们业务数据）：

    public interface MyCallable<T> {
        T call();
    }
1
2
3

下面是模板代码：

    protected <R> Observable<R> createObservable(final MyCallable<R> callable) {
        return Observable.create(new Observable.OnSubscribe<R>() {
            @Override
            public void call(Subscriber<? super R> subscriber) {
                try {
                    R result = callable.call();
                    subscriber.onNext(result);
                } catch (Exception e) {
                    subscriber.onError(e);
                }
                subscriber.onCompleted();
            }
        });
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14

使用就非常简单了调用createObservable方法，实现MyCallable接口即可，然后就是跟使用RxJava一样处理逻辑。

三、分析 GreenDao 是如何支持 RxJava 风格的

看过 GreenDao 源码的人知道，它也是通过这种方式支持 RxJava 的（下面看看他是怎么做的）：

    /**
     * Rx version of {@link AbstractDao#loadAll()} returning an Observable.
     */
    @Experimental
    public Observable<T> load(final K key) {
        return wrap(new Callable<T>() {
            @Override
            public T call() throws Exception {
                return dao.load(key);
            }
        });
    }
1
2
3
4
5
6
7
8
9
10
11
12

最终的实现也是通过 dao.load(key) 同步方法来实现的，关键是 wrap 方法了：

    protected <R> Observable<R> wrap(Callable<R> callable) {
        return wrap(RxUtils.fromCallable(callable));
    }

    //通过这个方法再包装了一层（就是默认设置执行的线程）
    protected <R> Observable<R> wrap(Observable<R> observable) {
        if (scheduler != null) {
            return observable.subscribeOn(scheduler);
        } else {
            return observable;
        }
    }

1
2
3
4
5
6
7
8
9
10
11
12
13

通过代码可以看到默认执行的线程是 IO 线程：

    /**
     * The returned RxDao is a special DAO that let's you interact with Rx Observables using RX's IO scheduler for
     * subscribeOn.
     *
     * @see #rxPlain()
     */
    @Experimental
    public RxDao<T, K> rx() {
        if (rxDao == null) {
            rxDao = new RxDao<>(this, Schedulers.io());
        }
        return rxDao;
    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14

所以使用 GreenDao 不用指定它在 IO 执行，因为框架已经帮我们设置了。

然后就是RxUtils.fromCallable(callable)方法了：

class RxUtils {
    /** As of RxJava 1.1.7, Observable.fromCallable is still @Beta, so just in case... */
    @Internal
    static <T> Observable<T> fromCallable(final Callable<T> callable) {
        return Observable.defer(new Func0<Observable<T>>() {

            @Override
            public Observable<T> call() {
                T result;
                try {
                    result = callable.call();
                } catch (Exception e) {
                    return Observable.error(e);
                }
                return Observable.just(result);
            }
        });
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

上面的注释说通过 Observable.fromCallable 也可以实现这样的逻辑，也就是说代替Observable.defer() 方法。
最后 GreenDao 是通过 defer 操作符来实现 RX 风格的。

四、defer和 create 操作符有什么异同点？

通过分析 GreenDao 源码得知，他是通过 defer 来做的，我们是通过 create 操作符来做的。那两者有什么不同？

我们对 defer 操作符比较陌生，先看看它的源码：

    public static <T> Observable<T> defer(Func0<Observable<T>> observableFactory) {
        return create(new OnSubscribeDefer<T>(observableFactory));
    }

1
2
3
4

说白了就是调用了create(OnSubscribe<T> f) 方法：

    public static <T> Observable<T> create(OnSubscribe<T> f) {
        return new Observable<T>(hook.onCreate(f));
    }
1
2
3

其实我们上面的 create 操作也是调用过来这个方法。只是 defer 操作符传递的 OnSubscribe 是 OnSubscribeDefer，那我们来看看这是什么鬼？

public final class OnSubscribeDefer<T> implements OnSubscribe<T> {
    final Func0<? extends Observable<? extends T>> observableFactory;

    public OnSubscribeDefer(Func0<? extends Observable<? extends T>> observableFactory) {
        this.observableFactory = observableFactory;
    }

    @Override
    public void call(final Subscriber<? super T> s) {
        Observable<? extends T> o;
        try {
            o = observableFactory.call();
        } catch (Throwable t) {
            Exceptions.throwOrReport(t, s);
            return;
        }
        o.unsafeSubscribe(Subscribers.wrap(s));
    }
    
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

OnSubscribeDefer 也是继承自 OnSubscribe，那么他的 call 方法肯定也是在订阅的时候被调用（就是说订阅的时候才创建这个 observable，并且每次订阅都会创建一个新的 observable）。
为什么 GreenDao 没有使用 create 那种方式精确控制数据的发射？现在 RxJava2.0 对create 操作符做出了一些限制，不能随随便便 create 了，这样出现一些问题。具体的RxJava2.0 的改动可以看看
他的 Github 说明 What’s-different-in-2.0

五、参考资料：

本文的例子放在 Github 上 https://github.com/chiclaim/AndroidAll

• pitfalls-of-operator-implementations
• subscribe vs unsafeSubscribe
• What’s-different-in-2.0

---

如果你觉得本文帮助到你，给我个关注和赞呗！

另外，我为 Android 程序员编写了一份：超详细的 Android 程序员所需要的技术栈思维导图。

如果有需要可以移步我的 GitHub -> AndroidAll，里面包含了最全的目录和对应知识点链接，帮你扫除 Android 知识点盲区。 由于篇幅原因只展示了 Android 思维导图：