千家信息网

请输入关键字词

热门搜索排行

最新搜索排行

导航： 首页 > 互联网科技 >

RxJava简单源码的示例分析

发表于：2026-11-03 作者：千家信息网编辑
千家信息网最后更新 2026年11月03日，今天就跟大家聊聊有关RxJava简单源码的示例分析，可能很多人都不太了解，为了让大家更加了解，小编给大家总结了以下内容，希望大家根据这篇文章可以有所收获。demo代码如下public class Ob
千家信息网最后更新 2026年11月03日RxJava简单源码的示例分析

今天就跟大家聊聊有关RxJava简单源码的示例分析，可能很多人都不太了解，为了让大家更加了解，小编给大家总结了以下内容，希望大家根据这篇文章可以有所收获。

demo代码如下

public class ObservableTest {    public static void main(String[] args) {        Observable observable = Observable.create(new ObservableOnSubscribe() {            @Override            public void subscribe(ObservableEmitter observer) throws Exception {                observer.onNext("处理的数字是：" + Math.random() * 100);                observer.onComplete();            }        });        observable.subscribe(new Consumer() {            @Override            public void accept(Object consumer) throws Exception {                System.out.println("我处理的元素是：" + consumer);            }        });    }}
先看第一行代码
Observable observable = Observable.create(new ObservableOnSubscribe() {    @Override    public void subscribe(ObservableEmitter observer) throws Exception {        observer.onNext("处理的数字是：" + Math.random() * 100);        observer.onComplete();    }});//Observable.java//第1560行public static  Observable create(ObservableOnSubscribe source) {    ObjectHelper.requireNonNull(source, "source is null");    //RxJavaPlugins里有很多方法可以设置，    //有点类似于Spring的ApplicationListener，在对应的生命周期中会被调用    return RxJavaPlugins.onAssembly(new ObservableCreate(source));}//RxJavaPlugins.java//第1031行public static  Observable onAssembly(@NonNull Observable source) {    Function f = onObservableAssembly;    //如果设置了对应的方法，就执行，否则原样返回    if (f != null) {        return apply(f, source);    }    return source;}
可以看到RxJavaPlugins中的方法如果不配置的方法，参数就会原样返回，所以Observable.create最终得到的就是ObservableCreate这个类。
再来看第二行代码
observable.subscribe(new Consumer() {    @Override    public void accept(Object consumer) throws Exception {        System.out.println("我处理的元素是：" + consumer);    }});//Observable.java//第10869行public final Disposable subscribe(Consumer onNext) {    return subscribe(onNext, Functions.ON_ERROR_MISSING, Functions.EMPTY_ACTION, Functions.emptyConsumer());}//Observable.java//第10958行public final Disposable subscribe(Consumer onNext, Consumer onError,            Action onComplete, Consumer onSubscribe) {    ObjectHelper.requireNonNull(onNext, "onNext is null");    ObjectHelper.requireNonNull(onError, "onError is null");    ObjectHelper.requireNonNull(onComplete, "onComplete is null");    ObjectHelper.requireNonNull(onSubscribe, "onSubscribe is null");    //这里的onNext就是我们自己写的Consumer类    LambdaObserver ls = new LambdaObserver(onNext, onError, onComplete, onSubscribe);    subscribe(ls);    return ls;}//Observable.java//第10974行public final void subscribe(Observer observer) {    ObjectHelper.requireNonNull(observer, "observer is null");    try {        observer = RxJavaPlugins.onSubscribe(this, observer);        ObjectHelper.requireNonNull(observer, "Plugin returned null Observer");        //还记得我们的observable变量是什么类型么？ObservableCreate！        subscribeActual(observer);    } catch (NullPointerException e) { // NOPMD        throw e;    } catch (Throwable e) {        Exceptions.throwIfFatal(e);        // can't call onError because no way to know if a Disposable has been set or not        // can't call onSubscribe because the call might have set a Subscription already        RxJavaPlugins.onError(e);        NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");        npe.initCause(e);        throw npe;    }}//ObservableCreate.java//第35行protected void subscribeActual(Observer observer) {    //这里的observer是LambdaObserver    CreateEmitter parent = new CreateEmitter(observer);    observer.onSubscribe(parent);    //省略部分代码}//LambdaObserver.java//第47行public void onSubscribe(Disposable s) {    //设置AtomicReference的值（LambdaObserver继承了AtomicReference）    //如果之前已经设置过了（AtomicReference的值不为空），则直接返回false    if (DisposableHelper.setOnce(this, s)) {        try {            //在new LambdaObserver()的时候我们设置了onSubscribe = Functions.emptyConsumer()            //所以这里什么都不做            onSubscribe.accept(this);        } catch (Throwable ex) {            Exceptions.throwIfFatal(ex);            s.dispose();            onError(ex);        }    }}//ObservableCreate.java//第35行protected void subscribeActual(Observer observer) {    //省略部分代码    try {        //还记得source是啥么，就是你在创建Observable的时候new的ObservableOnSubscribe        //于是终于执行到了我们编写的代码中        source.subscribe(parent);    } catch (Throwable ex) {        Exceptions.throwIfFatal(ex);        parent.onError(ex);    }}//ObservableOnSubscribe.java//第6行public static void main(String[] args) {    Observable observable = Observable.create(new ObservableOnSubscribe() {        //开始执行这个方法        //observer是new CreateEmitter(new LambdaObserver());        @Override        public void subscribe(ObservableEmitter observer) throws Exception {            observer.onNext("处理的数字是：" + Math.random() * 100);            observer.onComplete();        }    });}//ObservableCreate$CreateEmitter//第61行public void onNext(T t) {    if (t == null) {        onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));        return;    }    if (!isDisposed()) {        //这里的observer就是LambdaObserver        //t就是《"处理的数字是：" + Math.random() * 100》这段字符串        observer.onNext(t);    }}//LambdaObserver.java//第60行public void onNext(T t) {    if (!isDisposed()) {        try {            onNext.accept(t);        } catch (Throwable e) {            Exceptions.throwIfFatal(e);            get().dispose();            onError(e);        }    }}//ObservableOnSubscribe.java//第13行public static void main(String[] args) {    //省略部分代码        observable.subscribe(new Consumer() {        @Override        public void accept(Object consumer) throws Exception {            System.out.println("我处理的元素是：" + consumer);        }    });}//ObservableOnSubscribe.java//第8行public static void main(String[] args) {    Observable observable = Observable.create(new ObservableOnSubscribe() {        @Override        public void subscribe(ObservableEmitter observer) throws Exception {            //省略部分代码            observer.onComplete();        }    });    //省略部分代码}//ObservableCreate.java//第95行public void onComplete() {    if (!isDisposed()) {        try {            observer.onComplete();        } finally {            //取消订阅            dispose();        }    }}//LambdaObserver.java//第86行public void onComplete() {    if (!isDisposed()) {        lazySet(DisposableHelper.DISPOSED);        try {            //new LambdaObserver的时候设置了为空，所以不执行操作            onComplete.run();        } catch (Throwable e) {            Exceptions.throwIfFatal(e);            RxJavaPlugins.onError(e);        }    }}
至此，调用流程分析完成，可以看到虽然在main方法里我们只写了几行代码，但是内部调用的流程还是很繁杂的
看完上述内容，你们对RxJava简单源码的示例分析有进一步的了解吗？如果还想了解更多知识或者相关内容，请关注行业资讯频道，感谢大家的支持。


        


        
 很赞哦！ 

        


        



        代码
            处理
            方法
            就是
            部分
            数字
            分析
            元素
            内容
            时候
            源码
            示例
            原样
            流程
            繁杂
            一行
            参数
            变量
            周期
            字符
    
数据库的安全要保护哪些东西
数据库安全各自的含义是什么
生产安全数据库录入
数据库的安全性及管理
数据库安全策略包含哪些
海淀数据库安全审计系统
建立农村房屋安全信息数据库
易用的数据库客户端支持安全管理
连接数据库失败ssl安全错误
数据库的锁怎样保障安全

手磨咖啡
深圳九邦互联网科技有限公司
互联网科技产业
互联网巨头科技公司
互联网与科技金融数据
互联网科技未来
中国 科技 互联网
苏州大垚互联网科技
互联网科技股午盘股票涨跌
华中科技有哪些互联网大佬
中国互联网有什么科技含量
互联网科技寓意
米聚互联网科技南京有限公司排面
杭州披云加互联网科技有限公司
神州科技互联网有限公司
上海乐五互联网科技有限公司
艺电能源互联网科技有限公司招聘
美国五大互联网科技公司市值多少
互联网高科技绘画
中国主要互联网科技公司
互联网科技活动banner
上海致骋互联网科技
互联网科技的未来发展
广州恒生互联网科技有限
消费医药和互联网科技
长沙茂晟互联网科技有限公司
特约风豹互联网科技为什么扣费
中国科技和美国互联网
青岛易易互联网科技
现代科技的标志之一的互联网
        

        


        
        

        

          
        

      
    


    



    

      

        
相关文章

      

      
    






   





  



































0