RxJava 源码解析 —— Observable#defer(...)

摘要: 原创出处 http://www.iocoder.cn/RxJava/observable-defer/ 「芋道源码」欢迎转载，保留摘要，谢谢！

本文主要基于 RxJava 1.2.X 版本

本系列写作目的，为了辅助 Hystrix 的理解，因此会较为零散与琐碎，望见谅见谅。

在一些业务场景下，我们需要 Observable 是动态的，例如说，《Hystrix 源码解析 —— 执行结果缓存》分享的缓存 Observable ，无法在创建 Observable 阶段就知道是否有缓存，通过 Observable#defer(...) 方法，声明动态的 Observable 。示例代码如下：

public static void main(String[] args) {

Observable.defer(new Func0<Observable<String>>() { // #defer(...)

@Override

public Observable<String> call() {

String name = Math.random() > 0.5 ? "小明" : "小贾"; // 随机名字

return Observable.just(name);

}

}).subscribe(new Action1<String>() { // #subscribe(...)

@Override

public void call(String s) {

System.out.println(s);

}

});

}

Observable#defer(...) 方法，代码如下：

// Observable.java

public static <T> Observable<T> defer(Func0<Observable<T>> observableFactory) {

return create(new OnSubscribeDefer<T>(observableFactory));

}

public static <T> Observable<T> create(OnSubscribe<T> f) {

return new Observable<T>(RxJavaHooks.onCreate(f));

}

使用传入 observableFactory 参数，生成动态的 Observable 。

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));

}

在 Observable#subscribe(...) 方法调用时，调用 OnSubscribeDefer#call(...) 方法：
- 调用 Func0#call() 方法，创建动态的 Observable 。
- 调用 Observable#unsafeSubscribe(...) 方法，继续订阅逻辑。

RxJava 源码解析 —— Observable#defer(...)相关推荐

RxJava 源码解析之观察者模式
了解 RxJava 的应该都知道是一个基于事务驱动的库,响应式编程的典范.提到事务驱动和响应就不得不说说,设计模式中观察者模式,已经了解的朋友,可以直接跳过观察者模式的介绍,直接到 RxJava 源码 ...
rxJava源码解析系列五之Disposable
这个系列的前几篇文章是传统rxjava的数据流传递模式,我们知道,observable的链式调用,是下游被观察者持有上有被观察者的过程,发起订阅的时候是上游观察者持有下游观察者的过程.到最先的被观察者 ...
熔断器 Hystrix 源码解析 —— 命令执行（三）之执行超时
2019独角兽企业重金招聘Python工程师标准>>> 摘要: 原创出处 http://www.iocoder.cn/Hystrix/command-execute-third-ti ...
RXJava源码详解
rxjava Rx介绍 ReactiveX的历史 ReactiveX是Reactive Extensions的缩写,一般简写为Rx,最初是LINQ的一个扩展,由微软的架构师Erik Meijer领导的 ...
Rxjava源码分析之IO.Reactivex.Observable
Rxjava 源码系列目录 Rxjava源码分析之IO.Reactivex.Observer Rxjava源码分析之IO.Reactivex.CompositeDisposable Rxjava源码分 ...
Java熔断框架有哪些_降级熔断框架 Hystrix 源码解析：滑动窗口统计
降级熔断框架 Hystrix 源码解析:滑动窗口统计概述 Hystrix 是一个开源的降级熔断框架,用于提高服务可靠性,适用于依赖大量外部服务的业务系统.什么是降级熔断呢? 降级业务降级,是指牺牲 ...
RxPermissions 源码解析之举一反三
[toc] RxPermissions 源码解析简介 RxPermissions 是基于 RxJava 开发的用于帮助在Android 6.0 中处理运行时权限检测的框架.在 Android 6. ...
Retrofit源码解析
本文分析的源码基于retrofit:2.9.0. 前言 Retrofit基于OkHttp,网络请求的工作本质上是 OkHttp 完成,而 Retrofit 仅负责网络请求接口的封装,它们的关系示意图 ...
Colly源码解析——结合例子分析底层实现
通过<Colly源码解析--框架>分析,我们可以知道Colly执行的主要流程.本文将结合http://go-colly.org上的例子分析一些高级设置的底层实现.(转载请指明出于break ...

RxJava 源码解析 —— Observable#defer(...)

RxJava 源码解析 —— Observable#defer(...)相关推荐

最新文章

热门文章