RxJava 和 RxAndroid 五（线程调度）
对rxJava不了解的同学可以先看
RxJava 和 RxAndroid 一 (基础)
RxJava 和 RxAndroid 二（操作符的使用）
RxJava 和 RxAndroid 三（生命周期控制和内存优化）
RxJava 和 RxAndroid 四（RxBinding的使用）
本文将有几个例子说明，rxjava线程调度的正确使用姿势。
例1
Observable
.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
Logger.v( "rx_call" , Thread.currentThread().getName() );
subscriber.onNext( "dd");
subscriber.onCompleted();
}
})
.map(new Func1<String, String >() {
@Override
public String call(String s) {
Logger.v( "rx_map" , Thread.currentThread().getName() );
return s + "88";
}
})
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
Logger.v( "rx_subscribe" , Thread.currentThread().getName() );
}
}) ;
结果
/rx_call: main -- 主线程
/rx_map: main -- 主线程
/rx_subscribe: main -- 主线程
例2
new Thread(new Runnable() {
@Override
public void run() {
Logger.v( "rx_newThread" , Thread.currentThread().getName() );
rx();
}
}).start();
void rx(){
Observable
.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
Logger.v( "rx_call" , Thread.currentThread().getName() );
subscriber.onNext( "dd");
subscriber.onCompleted();
}
})
.map(new Func1<String, String >() {
@Override
public String call(String s) {
Logger.v( "rx_map" , Thread.currentThread().getName() );
return s + "88";
}
})
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
Logger.v( "rx_subscribe" , Thread.currentThread().getName() );
}
}) ;
}
结果
/rx_newThread: Thread-564 -- 子线程
/rx_call: Thread-564 -- 子线程
/rx_map: Thread-564 -- 子线程
/rx_subscribe: Thread-564 -- 子线程
- 通过例1和例2，说明，Rxjava默认运行在当前线程中。如果当前线程是子线程，则rxjava运行在子线程；同样，当前线程是主线程，则rxjava运行在主线程
例3
Observable
.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
Logger.v( "rx_call" , Thread.currentThread().getName() );
subscriber.onNext( "dd");
subscriber.onCompleted();
}
})
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.map(new Func1<String, String >() {
@Override
public String call(String s) {
Logger.v( "rx_map" , Thread.currentThread().getName() );
return s + "88";
}
})
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
Logger.v( "rx_subscribe" , Thread.currentThread().getName() );
}
}) ;
结果
/rx_call: RxCachedThreadScheduler-1 --io线程
/rx_map: main --主线程
/rx_subscribe: main --主线程
例4
Observable
.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
Logger.v( "rx_call" , Thread.currentThread().getName() );
subscriber.onNext( "dd");
subscriber.onCompleted();
}
})
.map(new Func1<String, String >() {
@Override
public String call(String s) {
Logger.v( "rx_map" , Thread.currentThread().getName() );
return s + "88";
}
})
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
Logger.v( "rx_subscribe" , Thread.currentThread().getName() );
}
}) ; 结果
/rx_call: RxCachedThreadScheduler-1 --io线程
/rx_map: RxCachedThreadScheduler-1 --io线程
/rx_subscribe: main --主线程
- 通过例3、例4 可以看出 .subscribeOn(Schedulers.io()) 和 .observeOn(AndroidSchedulers.mainThread()) 写的位置不一样，造成的结果也不一样。从例4中可以看出 map() 操作符默认运行在事件产生的线程之中。事件消费只是在 subscribe（） 里面。
- 对于 create() , just() , from() 等 --- 事件产生
map() , flapMap() , scan() , filter() 等 -- 事件加工
subscribe() -- 事件消费
- 事件产生：默认运行在当前线程，可以由 subscribeOn() 自定义线程
事件加工：默认跟事件产生的线程保持一致, 可以由 observeOn() 自定义线程
事件消费：默认运行在当前线程，可以有observeOn() 自定义
例5 多次切换线程
Observable
.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
Logger.v( "rx_call" , Thread.currentThread().getName() );
subscriber.onNext( "dd");
subscriber.onCompleted();
}
})
.observeOn( Schedulers.newThread() ) //新线程
.map(new Func1<String, String >() {
@Override
public String call(String s) {
Logger.v( "rx_map" , Thread.currentThread().getName() );
return s + "88";
}
})
.observeOn( Schedulers.io() ) //io线程
.filter(new Func1<String, Boolean>() {
@Override
public Boolean call(String s) {
Logger.v( "rx_filter" , Thread.currentThread().getName() );
return s != null ;
}
})
.subscribeOn(Schedulers.io()) //定义事件产生线程：io线程
.observeOn(AndroidSchedulers.mainThread()) //事件消费线程：主线程
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
Logger.v( "rx_subscribe" , Thread.currentThread().getName() );
}
}) ;结果
/rx_call: RxCachedThreadScheduler-1 -- io 线程
/rx_map: RxNewThreadScheduler-1 -- new出来的线程
/rx_filter: RxCachedThreadScheduler-2 -- io线程
/rx_subscribe: main -- 主线程
例6：只规定了事件产生的线程
Observable
.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
Log.v( "rx--create " , Thread.currentThread().getName() ) ;
subscriber.onNext( "dd" ) ;
}
})
.subscribeOn(Schedulers.io())
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
Log.v( "rx--subscribe " , Thread.currentThread().getName() ) ;
}
}) ;
结果
/rx--create: RxCachedThreadScheduler-4 // io 线程
/rx--subscribe: RxCachedThreadScheduler-4 // io 线程
例:7：只规定事件消费线程
Observable
.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
Log.v( "rx--create " , Thread.currentThread().getName() ) ;
subscriber.onNext( "dd" ) ;
}
})
.observeOn( Schedulers.newThread() )
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
Log.v( "rx--subscribe " , Thread.currentThread().getName() ) ;
}
}) ;
结果
/rx--create: main -- 主线程
/rx--subscribe: RxNewThreadScheduler-1 -- new 出来的子线程
从例6可以看出，如果只规定了事件产生的线程，那么事件消费线程将跟随事件产生线程。
从例7可以看出，如果只规定了事件消费的线程，那么事件产生的线程和 当前线程保持一致。
例8：线程调度封装
在Android 常常有这样的场景，后台处理处理数据，前台展示数据。
一般的用法：
Observable
.just( "123" )
.subscribeOn( Schedulers.io())
.observeOn( AndroidSchedulers.mainThread() )
.subscribe(new Action1() {
@Override
public void call(Object o) {
}
}) ;
但是项目中这种场景有很多，所以我们就想能不能把这种场景的调度方式封装起来，方便调用。
简单的封装
public Observable apply( Observable observable ){
return observable.subscribeOn( Schedulers.io() )
.observeOn( AndroidSchedulers.mainThread() ) ;
}
使用
apply( Observable.just( "123" ) )
.subscribe(new Action1() {
@Override
public void call(Object o) {
}
}) ;
弊端：虽然上面的这种封装可以做到线程调度的目的，但是它破坏了链式编程的结构，是编程风格变得不优雅。
改进：Transformers 的使用（就是转化器的意思，把一种类型的Observable转换成另一种类型的Observable ）
改进后的封装
Observable.Transformer schedulersTransformer = new Observable.Transformer() {
@Override public Object call(Object observable) {
return ((Observable) observable).subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread());
}
};
使用
Observable
.just( "123" )
.compose( schedulersTransformer )
.subscribe(new Action1() {
@Override
public void call(Object o) {
}
}) ;
弊端：虽然保持了链式编程结构的完整，但是每次调用 .compose( schedulersTransformer ) 都是 new 了一个对象的。所以我们需要再次封装，尽量保证单例的模式。
改进后的封装
package lib.app.com.myapplication;
import rx.Observable;
import rx.android.schedulers.AndroidSchedulers;
import rx.schedulers.Schedulers;
/**
* Created by ${zyj} on 2016/7/1.
*/
public class RxUtil {
private final static Observable.Transformer schedulersTransformer = new Observable.Transformer() {
@Override public Object call(Object observable) {
return ((Observable) observable).subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread());
}
};
public static <T> Observable.Transformer<T, T> applySchedulers() {
return (Observable.Transformer<T, T>) schedulersTransformer;
}
}
使用
Observable
.just( "123" )
.compose( RxUtil.<String>applySchedulers() )
.subscribe(new Action1() {
@Override
public void call(Object o) {
}
}) ;
RxJava 和 RxAndroid 五（线程调度）相关推荐
- 重拾Android之路(五)RxJava和RxAndroid
现在RxJava和RxAndroid越来越火爆,自己在业余时间也学习了一下,感觉确实很好用,之前 为了完成页面刷新,数据请求,组件信息传递的时候,要使用handler,真的是逻辑思路很强,稍微不注意, ...
- RxJava 和 RxAndroid 三（生命周期控制和内存优化）
前言:对Rxjava.Rxandroid不了解的同学可以先看看 RxJava 和 RxAndroid RxJava 和 RxAndroid 二(操作符的使用) RxJava使我们很方便的使用链式编程, ...
- RxJava和RxAndroid
现在RxJava和RxAndroid越来越火爆,自己在业余时间也学习了一下,感觉确实很好用,之前 为了完成页面刷新,数据请求,组件信息传递的时候,要使用handler,真的是逻辑思路很强,稍微不注意, ...
- 这样，你才能算会 RxJava、RxAndroid 之取消订阅
一:概述 前几天app 总是空指针奔溃,发现为某个 Fragment 中的控件为null 导致的.而且这个Fragment 是能够正常显示的.那么为什么还会空指针呢? 二:问题排查 这个app使用了 ...
- RxJava和RxAndroid学习记录
目录 1 概念和说明 1.1 响应式编程 1.2 RxJava 1.3 关于RxJava和RxAndroid 1.4 关于响应式编程和普通编程 2. 基本使用 2.1 基本元素关系图 2.2 代码示例 ...
- RxJava 和 RxAndroid 一 (基础)
1.RxJava 项目地址 https://github.com/ReactiveX/RxJava 2.RxAndroid 项目地址 https://github.com/ReactiveX/R ...
- RxJava 和 RxAndroid 二（操作符的使用）
前言:对Rx不了解的朋友可以先看我的第一篇博文 RxJava 和 RxAndroid 一 (基础),是对Rxjava的基本介绍 1.merge操作符,合并观察对象 19 List<String ...
- android RxJava（RxAndroid）的简单使用
今天,简单讲讲android里如何使用RxJava(RxAndroid). Android框架系列: 一.android EventBus的简单使用 二.android Glide简单使用 三.and ...
- RxJava 和 RxAndroid 四（RxBinding的使用）
对Rxjava不熟悉的同学可以先看我之前写的几篇文章 RxJava 和 RxAndroid 一 (基础) RxJava 和 RxAndroid 二(操作符的使用) RxJava 和 RxAndroid ...
最新文章
- Mac环境下mysql安装以及登录
- Delphi编译指令说明
- Android 蓝牙操作－－读取远程已配对的蓝牙设备
- 浅谈微信小程序对于房地产行业的影响
- 最大和 -- 最大子矩阵
- 全部物理宇宙全部能由数学理解
- 互联网晚报 | 3月28日 星期一 | 上海医保局辟谣：新冠患者要自负医疗费用不属实 ；十荟团关停全国所有业务...
- 零元学Expression Blend 4 - Chapter 3 熟悉操作第一步(制作一个猴子脸)
- php中调行高代码_网页代码如何调整行距
- puppet的相关介绍
- android studio或者IntelliJ代码样式的设置
- java设置登录超时时间设置_session超时时间设置方法
- Activiti 7 配置及相关流程一站式介绍（包含基础流程及网关应用）
- java overload_java之方法重载（overload）
- oc引导windows蓝屏_最全的Windows 系统蓝屏死机代码的含义及解决方案
- 【读书笔记】你离考研成功就差这本书
- java全局搜素快捷键_eclipse全局搜索快捷键是什么
- windows下载安装配置Geany
- 求两个三维向量的夹角（带正负）
- FPGA中高时钟频率计数器设计
热门文章
- libsvm matlab 调参数,matlab – 使用libsvm调整svm
- 研究生一年级，非计算机专业，自学机器学习现实吗？
- 来试试读论文的新神器！AMiner发布“论文背景文献”一键生成工具，帮你搞清一篇论文的“来龙去脉”...
- 开箱即用！这个神器，拯救了无数算法工程师……
- 阿里广告技术最新突破：全链路联动-面向最终目标的全链路一致性建模
- OpenViDial：一个大规模多模态对话数据集
- ACM MM 2020视频目标检测挑战赛冠军DeepBlueAI团队技术分享
- 开放域知识库问答研究回顾
- 第三届“中国法研杯”司法人工智能挑赛（CAIL2020）即将启动
- 数学建模第四节2020.4.24-5.3补