系列文章

RxJava系列之简介和观察者设计模式
RxJava系列之上游与下游
RxJava系列之常用创建型操作符
RxJava系列之常用变换操作符
RxJava系列之常用过滤操作符
RxJava系列之常用条件操作符
RxJava系列之常用合并操作符
RxJava系列之常用异常操作符
RxJava系列之线程切换实战
RxJava系列之背压模式
RxJava系列之配合Retrofit
RxJava系列之泛型高级
RxJava系列之手写create操作符
RxJava系列之手写create操作符增加泛型限定
RxJava系列之手写just操作符
RxJava系列之手写map操作符
RxJava系列之手写切换线程

什么是上游和下游

在RxJava中，我们将事件的起源称之为上游，事件的处理称之为下游。上游产生事件之后，发射给下游，下游接收到并请求。如下图所示：

简单例子

/*** 流程整理 1* @param vieww*/public void r04(View vieww) {// 上游 Observable 被观察者Observable.create(new ObservableOnSubscribe<String>() {@Overridepublic void subscribe(ObservableEmitter<String> emitter) throws Exception {// 发射Log.d(TAG, "上游 subscribe: 开始发射..."); // todo 2emitter.onNext("RxJavaStudy");emitter.onComplete(); // 发射完成  // todo 4// 上游的最后log才会打印Log.d(TAG, "上游 subscribe: 发射完成");}}).subscribe(// 下游 Observer 观察者new Observer<String>() {@Overridepublic void onSubscribe(Disposable d) {// 弹出 加载框 ....Log.d(TAG, "上游和下游订阅成功 onSubscribe 1"); // todo 1}@Overridepublic void onNext(String s) {Log.d(TAG, "下游接收 onNext: " + s); // todo 3}@Overridepublic void onError(Throwable e) {}@Overridepublic void onComplete() {// 隐藏加载框Log.d(TAG, "下游接收完成 onComplete"); // todo 5  只有接收完成之后，上游的最后log才会打印}});/***  D/MainActivity: 上游和下游订阅成功 onSubscribe 1*  D/MainActivity: 上游 subscribe: 开始发射...*  D/MainActivity: 下游接收 onNext: RxJavaStudy*  D/MainActivity: 下游接收完成 onComplete*  D/MainActivity: 上游 subscribe: 发射完成*/}

整个流程是，上下游订阅成功，下游收到通知，处理订阅成功事件。上游在订阅成功的回调中发射事件，下游收到onNext回调，处理上游发送的事件。上游发送完成事件，下游收到onComplete回调。

另外一个例子

  /*** 流程整理2* @param vieww*/public void r05(View vieww) {// 上游 Observable 被观察者Observable.create(new ObservableOnSubscribe<String>() {@Overridepublic void subscribe(ObservableEmitter<String> emitter) throws Exception {// 发射Log.d(TAG, "上游 subscribe: 开始发射..."); // todo 2emitter.onNext("RxJavaStudy");// emitter.onComplete(); // 发射完成  // todo 4// 上游的最后log才会打印// Log.d(TAG, "上游 subscribe: 发射完成");// emitter.onError(new IllegalAccessException("error rxJava"));// TODO 结论：在 onComplete();/onError 发射完成 之后 再发射事件  下游不再接收上游的事件/*emitter.onNext("a");emitter.onNext("b");emitter.onNext("c");*/// 发一百个事件emitter.onError(new IllegalAccessException("error rxJava")); // 发射错误事件emitter.onComplete(); // 发射完成// TODO 结论：已经发射了onComplete();， 再发射onError RxJava会报错，不允许// TODO 结论：先发射onError，再onComplete();，不会报错， 有问题（onComplete不会接收到了）}}).subscribe(// 下游 Observer 观察者new Observer<String>() {@Overridepublic void onSubscribe(Disposable d) {// 弹出 加载框 ....Log.d(TAG, "上游和下游订阅成功 onSubscribe 1"); // todo 1}@Overridepublic void onNext(String s) {Log.d(TAG, "下游接收 onNext: " + s); // todo 3}@Overridepublic void onError(Throwable e) {Log.d(TAG, "下游接收 onError: " + e.getMessage());}@Overridepublic void onComplete() {// 隐藏加载框Log.d(TAG, "下游接收完成 onComplete"); // todo 5  只有接收完成之后，上游的最后log才会打印}});}

在 onComplete();/onError 发射完成之后再发射事件下游不再接收上游的事件
已经发射了onComplete();，再发射onError RxJava会报错，不允许
已经发射了onComplete();，再发射onError RxJava会报错，不允许

中断事件

Disposable d;/*** 切断下游，让下游不再接收上游的事件，也就是说不会去更新UI* @param view*/public void r06(View view) {// TODO 上游 ObservableObservable.create(new ObservableOnSubscribe<Integer>() {@Overridepublic void subscribe(ObservableEmitter<Integer> e) throws Exception {for (int i = 0; i < 100; i++) {e.onNext(i);}e.onComplete();}})// 订阅下游.subscribe(new Observer<Integer>() {@Overridepublic void onSubscribe(Disposable d) {MainActivity.this.d = d;}@Overridepublic void onNext(Integer integer) {Log.d(TAG, "下游接收 onNext: " + integer);// 接收上游的一个事件之后，就切断下游，让下游不再接收// d.dispose();}@Overridepublic void onError(Throwable e) {}@Overridepublic void onComplete() {}});}@Overrideprotected void onDestroy() {super.onDestroy();// 切断下游if (d != null) d.dispose();}

下游在订阅成功之后onSubscribe(Disposable d) 回调方法传递过来Disposable对象，可以通过这个对象可以切断下游。

RxJava系列之上游与下游相关推荐

RxJava基本使用--上游与下游
RxJava基本使用–上游与下游 var disposable : Disposable? = nullprivate fun rxTest1() {//起点被观察者Observable.creat ...
MKS上游和下游集成式压力控制器的技术分析及其替代解决方案
摘要:目前的MKS系列集成式压力控制器本质上是一种流量调节和测量装置,无法直接用来进行准确的压力控制,而且MKS压力控制器还存在测量精度不高.压力控制范围有限和对工作介质洁净度要求很高的不足.为此,为 ...
RxJava系列6(从微观角度解读RxJava源码)
RxJava系列1(简介) RxJava系列2(基本概念及使用介绍) RxJava系列3(转换操作符) RxJava系列4(过滤操作符) RxJava系列5(组合操作符) RxJava系列6(从微观角 ...
RxJava系列四（过滤操作符）
转载请注明出处:https://zhuanlan.zhihu.com/p/21966621 RxJava系列1(简介) RxJava系列2(基本概念及使用介绍) RxJava系列3(转换操作符) Rx ...
python 处理snp的vcf文件，统计snp在基因的intron、exon还是上游、下游还是不在基因及基因附近
1.c处理vcf文件,初步统计snp位置位置信息有5种 down_stream是gene下游2k以内,up_stream 是gene上游2k以内,gene_exon是snp在外显子内,gene_in ...
软件开发设计中的上游与下游
生产流程中的上下游让我们以一个简单的生产流程开始,尽管它跟软件开发没有关系,这样我们能以此为基础定义软件开发中的上下游. 上面的例子有三个步骤:收集部件.组装部件.喷漆. 一个生产流程跟一条河流很相 ...
RxJava系列7(最佳实践)
RxJava系列1(简介) RxJava系列2(基本概念及使用介绍) RxJava系列3(转换操作符) RxJava系列4(过滤操作符) RxJava系列5(组合操作符) RxJava系列6(从微观角 ...
RxJava系列1(简介)
RxJava系列1(简介) RxJava系列2(基本概念及使用介绍) RxJava系列3(转换操作符) RxJava系列4(过滤操作符) RxJava系列5(组合操作符) RxJava系列6(从微观角 ...
软件开发中的上游和下游
听了腾讯犀牛鸟王佳对开源的讲解中,对其中的专业术语上游(upstream)和下游(downstream)不理解,看了这篇blog理解了特此转载记录. 看了以下blog后个人理解:自然界上水是高处往低处 ...

RxJava系列之上游与下游