Android异步框架 RxJava
观察者模式的概念
观察者a与被观察者b建立订阅关系,当被观察者b发生某种改变时,立即通知观察者a
添加依赖
compile 'io.reactivex.rxjava2:rxjava:2.1.0' compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
基本模式
observable被观察者
注意各地方添加泛型避免大片警告,onnext()是事件的回调,oncomplete()是事件的结尾。oncomplete()与onerror互斥需要保持唯一性,并只能调用一次。
observable<string> observable= observable.create(new observableonsubscribe<string>() { @override public void subscribe(observableemitter<string> e) throws exception { e.onnext("消息1"); e.onnext("消息2"); e.onnext("消息3"); e.oncomplete(); } });
observer观察者
创建观察者时回调的onsubscribe可以获取disposable对象,在合适的时候判断条件,调用dispose()即可接触订阅关系
observer<string> observer=new observer<string>() { @override public void onsubscribe(disposable d) { //通过判断解除订阅关系 d.dispose(); } @override public void onnext(string o) { //对应observable的onnext方法 } @override public void onerror(throwable e) { //对应observable的onerror方法 } @override public void oncomplete() { //对应observable的oncomplete方法 } };
建立订阅关系
observable.subscribeon(schedulers.io()) //指定事件生产在子线程 .observeon(androidschedulers.mainthread()) //指定事件消费在ui线程 .subscribe(observer);
observable被观察者的其他模式
//just模式,将自动发送onnext()事件 observable<string> observable = observable.just("发送消息"); //fromiterable模式,遍历集合,并自动发送onnext()事件 observable<string> observable = observable.fromiterable((iterable<string>) mlist); //interval模式,定时自动发送整数序列,从0开始每隔2秒计数, observable<long> observable = observable.interval(0,2, timeunit.seconds) //range模式,自动发送特定的整数序列,0表示不发送,负数会抛异常,从1开始发送到20 observable<integer> observable = observable.range(1,20); //timer模式,定时执行观察者的onnext()方法 observable<integer> observable = observable.timer(2, timeunit.seconds);
observable被观察者的更多创建方式以及操作符
如创建操作,数据过滤操作,条件操作,转载以下博客,很详细:
rxjava操作符大全
scheduler调度器
四种常见模式
schedulers.immediate() 默认模式,在当前线程运行
schedulers.newthread() 创建新的子线程运行
schedulers.io() 创建新的子线程运行,内部使用的是无上限的线程池,可重用空闲的线程,效率高
androidschedulers.mainthread() 在ui主线程运行
订阅事件时的生产与消费线程
subscribeon() 指定observable(被观察者)所在的线程,或者叫做事件产生的线程
observeon() 指定 observer(观察者)所运行在的线程,或者叫做事件消费的线程
新的观察者模式
flowable被观察者
flowable<string> flowable = flowable.create(new flowableonsubscribe<string>() { @override public void subscribe(flowableemitter<string> e) throws exception { e.onnext("hello rxjava!"); e.oncomplete(); } },backpressurestrategy.buffer);//增加背压模式
subscriber观察者
onsubscribe()会返回subscription对象,调用cancel()即可取消订阅关系,request()即可指定消费事件的数量
subscriber<string> subscriber=new subscriber<string>() { @override public void onsubscribe(subscription s) { s.request(long.max_value); } @override public void onnext(string s) { log.i("rxjava", "onnext: "+s); } @override public void onerror(throwable t) { log.i("rxjava", "onerror"); } @override public void oncomplete() { log.i("rxjava", "oncomplete"); } }; flowable.subscribe(subscriber);//建立订阅关系
backpressure背压模式
如果生产者和消费者不在同一线程的情况下,如果生产者的速度大于消费者的速度,就会产生backpressure问题。即异步情况下,backpressure问题才会存在。
buffer
这样,消费者通过request()即使传入一个很大的数字,生产者也会生产事件,并将处理不了的事件缓存。
但是这种方式任然比较消耗内存,除非是我们比较了解消费者的消费能力,能够把握具体情况,不会产生oom。
drop
当消费者处理不了事件,就丢弃。
消费者通过request()传入其需求n,然后生产者把n个事件传递给消费者供其消费。其他消费不掉的事件就丢掉
latest
latest与drop功能基本一致,唯一的区别就是latest总能使消费者能够接收到生产者产生的最后一个事件
error
这种方式会在产生backpressure问题的时候直接抛出一个异常,这个异常就是著名的missingbackpressureexception
上一篇: JedisPool工具类
推荐阅读
-
Android异步框架 RxJava
-
一款好用的android图片缩放查看开源框架photoview的使用。
-
Android极封:Retrofit2+RxJava网络请求框架
-
Android-第三方框架-网络请求之okhttp
-
Android视频播放之JiaoZiVideoPlayer框架的使用
-
YII2 框架 如何 接收 支付宝异步通知
-
apache mina (异步连接框架)介绍 MinaApache框架log4jSocket
-
android在异步任务中关闭Cursor的代码方法
-
Android自动化测试实战教程(Java篇+主流工具+框架+脚本) android框架软件测试
-
android存储访问框架Storage Access Framework