使用RxJava可以方便实现观察者模式，数据转换和线程之间通信
https://github.com/ReactiveX/RxJava
在Android中使用RxAndroid增加安卓主线程支持：
https://github.com/ReactiveX/RxAndroid
实际开发中在gradle中注册对应的依赖即可：
compile 'io.reactivex:rxandroid:1.1.0'
观察者模式一般要有两个对象：
1.被观察者：Obserable
2.观察者：Observer
观察者和被观察者绑定后，被观察者在某种事件发生时会向所有观察它的观察者发送事件，即调用观察者的回调函数。
使用java.util里的观察者应该是这样的：
import android.util.Log;
import java.util.Observable;
import java.util.Observer;
public class ObserverTest {
void test0() {
//创建被观察者
DemoObserable obserable = new DemoObserable();
//创建一个观察者
Observer observer1 = new Observer() {
//观察者接收到事件时的操作
@Override
public void update(Observable observable, Object data) {
Log.e("TAG", "ToObserver1: " + data);
}
};
//添加绑定
obserable.addObserver(observer1);
//被观察者发送事件
obserable.subscribe("Hello World!!");
//被观察者将观察者移除
obserable.deleteObserver(observer1);
}
//被监听者
public class DemoObserable extends Observable {
//被监听者发送数据给所有观察者
void subscribe(String data) {
setChanged();
notifyObservers(data);
}
}
}
假设观察者有多个，可以依次注册，删除：
//创建被观察者
DemoObserable obserable = new DemoObserable();
//创建一个观察者
Observer observer1 = new Observer() {
//观察者接收到事件时的操作
@Override
public void update(Observable observable, Object data) {
Log.e("TAG", "ToObserver1: " + data);
}
};
//创建一个观察者
Observer observer2 = new Observer() {
//观察者接收到事件时的操作
@Override
public void update(Observable observable, Object data) {
Log.e("TAG", "ToObserver2: " + data);
}
};
//添加绑定
obserable.addObserver(observer1);
obserable.addObserver(observer2);
然而，我们接下来使用 rx.Observer, rx.Observable, 思想也是类似的：
1.观察者与被观察者
void test0() {
Observable.OnSubscribe<String> onSubscribe = new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("hello world!!");
subscriber.onCompleted();
}
};
Observable<String> observable = Observable.create(onSubscribe);
Observer<String> observer = new Observer<String>() {
@Override
public void onNext(String s) {
Log.e("TAG", s);
}
@Override
public void onCompleted() {
Log.e("TAG", "completed");
}
@Override
public void onError(Throwable e) {
Log.e("TAG", "error" + e);
}
};
observable.subscribe(observer);
}
完成了类似的功能。
Observable构造方法是保护的，只能通过提供的静态方法创建对象，如Observable.create()
RxJava可以使用链式调用简化代码，所以也可以写成：
Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("hello world!!");
subscriber.onCompleted();
}
}).subscribe(new Observer<String>() {
@Override
public void onNext(String s) {
Log.e("TAG", s);
}
@Override
public void onCompleted() {
Log.e("TAG", "completed");
}
@Override
public void onError(Throwable e) {
Log.e("TAG", "error" + e);
}
});
如果直接知道被监听者发送事件onNext()时的参数，可以使用just()来创建Observable，效果和上例一样
Observable.just("Hello World!")
.subscribe(new Observer<String>() {
@Override
public void onCompleted() {
Log.e("TAG", "complete");
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(String s) {
Log.e("TAG", s);
}
});
just函数可以传入多个参数，事件依次调用后最后执行onComplete()
Observable.just("Hello World!", "RxJava Demo")
.subscribe(new Observer<String>() {
@Override
public void onCompleted() {
Log.e("TAG", "complete");
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(String s) {
Log.e("TAG", s);
}
});
如果观察者只需要重写onNext()方法，可以直接用ActionX来节省掉onError和onComplete的代码：
Observable.just("Hello World!", "RxJava Demo")
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
Log.e("TAG", s);
}
});
现在观察者也变的好简单，使用多个参数just()来创建也可以用from(T[])来创建，与上例等价：
String[] arr = {"Hello World!", "RxJava Demo"};
Observable.from(arr)
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
Log.e("TAG", s);
}
});
如果不想用Observer，又想处理onError和onComplete，就需要在subscribe()函数传多个ActionX：
String[] arr = {"Hello World!", "RxJava Demo"};
Action1 nextAction, errorAction;
Action0 completeAction;
nextAction = new Action1<String>() {
@Override
public void call(String s) {
Log.e("TAG", s);
}
};
errorAction = new Action1<Exception>() {
@Override
public void call(Exception e) {
e.printStackTrace();
}
};
completeAction = new Action0() {
@Override
public void call() {
Log.e("TAG", "complete");
}
};
Observable.from(arr)
.subscribe(nextAction, errorAction, completeAction);
2.观察者与被观察者在不同线程
加上线程控制语句subscribeOn，observeOn，就可以将被观察者和观察者指定在不同线程中：
Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
Log.e("TAG", "I am in computation");
subscriber.onNext("Hello World!!");
}
})
.subscribeOn(Schedulers.computation())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
Log.e("TAG", s);
}
});
两个操作发生在不同线程中，通过log可以看出来：
03-29 21:00:39.333 13153-13226/com.hzy.rxjavademo E/TAG: I am in computation
03-29 21:00:39.334 13153-13153/com.hzy.rxjavademo E/TAG: Hello World!!
03-29 21:00:39.914 13153-13227/com.hzy.rxjavademo E/TAG: I am in computation
03-29 21:00:39.915 13153-13153/com.hzy.rxjavademo E/TAG: Hello World!!
03-29 21:00:40.607 13153-13228/com.hzy.rxjavademo E/TAG: I am in computation
03-29 21:00:40.608 13153-13153/com.hzy.rxjavademo E/TAG: Hello World!!
通过日志pid可以看到，Observable执行会开启新的线程，而Observer观察者收到消息后执行的操作在UI线程执行，所以可以方便的替代之前Android的new Thread加Handler发送消息的机制。
如下代码就可以完成一个完整的异步网络请求，并把结果和出错信息通知到UI线程的观察者：
Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
try {
URL url = new URL("https://m.baidu.com/");
InputStream is = url.openStream();
InputStreamReader isr = new InputStreamReader(is);
BufferedReader br = new BufferedReader(isr);
String line;
StringBuilder builder = new StringBuilder();
while ((line = br.readLine()) != null) {
builder.append(line);
}
subscriber.onNext(builder.toString());
} catch (Exception e) {
subscriber.onError(e);
}
}
})
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
Log.e("TAG", s);
}
}, new Action1<Throwable>() {
@Override
public void call(Throwable throwable) {
throwable.printStackTrace();
}
});
异步操作变得好简单，而且很容易理解
3.数据的过滤
数据过滤使用filter方法：
String[] arr = {"Hello World!", "RxJava Demo", "RxAndroid"};
Observable.from(arr)
.filter(new Func1<String, Boolean>() {
@Override
public Boolean call(String s) {
return s.startsWith("Rx");
}
})
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
Log.e("TAG", s);
}
});
这时，如例子中所有返回以Rx开头字符串的事件，才会去通知观察者。
4.数据的转换
消息发送者返回的数据用map()进行转化后再发送给观察者
Observable.just(123)
.map(new Func1<Integer, String>() {
@Override
public String call(Integer integer) {
String ret = integer + ":" + integer * 2;
return ret;
}
}).subscribe(new Action1<String>() {
@Override
public void call(String s) {
Log.e("TAG", s);
}
});
把消息发送者发来的 Integer类型转成String类型再传给观察者去处理。
使用lift()把一个观察者转成另一个：
Observable.just(123)
.lift(new Observable.Operator<String, Integer>() {
@Override
public Subscriber<? super Integer> call(final Subscriber<? super String> subscriber) {
return new Subscriber<Integer>() {
@Override
public void onCompleted() {
subscriber.onCompleted();
}
@Override
public void onError(Throwable e) {
subscriber.onError(e);
}
@Override
public void onNext(Integer integer) {
subscriber.onNext("hello" + integer);
}
};
}
}).subscribe(new Action1<String>() {
@Override
public void call(String s) {
Log.e("TAG", s);
}
});
把一个Integer类型的Subscriber转成String类型，Subscriber实现Observer接口，所以
Subscriber也是观察者。
使用compose直接把被监听者转换掉，Transformer规定了转换的规则，Obserable怎么转换呢，还是用个map来转吧，Transformer在此显得多此一举了：
Observable.just(123)
.compose(new Observable.Transformer<Integer, String>() {
@Override
public Observable<String> call(Observable<Integer> integerObservable) {
return integerObservable.map(new Func1<Integer, String>() {
@Override
public String call(Integer integer) {
return integer + "hello";
}
});
}
}).subscribe(new Action1<String>() {
@Override
public void call(String s) {
Log.e("TAG", s);
}
});
把Integer类型转换成String类型
5.条件判断
条件判断的Observable范型都会被转换成Boolean类型
判断是否存在某种类型用exists()函数
Observable.just(123, 20, 34, 4, 0, 2)
.exists(new Func1<Integer, Boolean>() {
@Override
public Boolean call(Integer integer) {
return integer == 0;
}
})
.subscribe(new Action1<Boolean>() {
@Override
public void call(Boolean aBoolean) {
Log.e("TAG", "" + aBoolean);
}
});
判断那些值中是否存在为0的Integer,显然存在，返回true
用all()判断是否全都是某种情况：
Observable.just(123, 20, 34, 4, 0, 2)
.all(new Func1<Integer, Boolean>() {
@Override
public Boolean call(Integer integer) {
return integer > 0;
}
})
.subscribe(new Action1<Boolean>() {
@Override
public void call(Boolean aBoolean) {
Log.e("TAG", "" + aBoolean);
}
});
上面那些值全都大于0么，显然不是，返回false。
6.观察者嵌套
有一种场景，观察者收到消息后，不能完全处理掉消息，自身又作为消息发送者给其他观察者发消息。
Observable.just(123)
.subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Observable.just(integer)
.subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.e("TAG", "" + integer);
}
});
}
});
两层嵌套之后缩进已经很长了，为解决这个问题，可以用flatMap()来减少缩进：
Observable.just(123)
.flatMap(new Func1<Integer, Observable<Integer>>() {
@Override
public Observable<Integer> call(Integer integer) {
return Observable.just(integer);
}
}).subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.e("TAG", "" + integer);
}
});
这样，缩进好多了。
最后是一个综合实例：
//使用IO线程读取SD卡上所有png文件并在主线程上打印
private void testy() {
Observable.create(new Observable.OnSubscribe<File>() {
@Override
public void call(Subscriber<? super File> subscriber) {
//获取外部存储上的所有文件
File exDir = Environment.getExternalStorageDirectory();
traceFiles(exDir, subscriber);
}
}).subscribeOn(Schedulers.io())
//过滤出png文件
.filter(new Func1<File, Boolean>() {
@Override
public Boolean call(File file) {
return file.isFile() && file.getName().endsWith(".png");
}
//使用buffer防止消息发送过快
}).onBackpressureBuffer()
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Action1<File>() {
@Override
public void call(File file) {
Log.e("TAG", file.getAbsolutePath());
}
}, new Action1<Throwable>() {
@Override
public void call(Throwable throwable) {
Log.e("TAG", "Error:" + throwable);
throwable.printStackTrace();
}
});
}
private void traceFiles(File file, Subscriber<? super File> subscriber) {
subscriber.onNext(file);
if (file.isDirectory()) {
for (File f : file.listFiles()) {
traceFiles(f, subscriber);
}
}
}
结束
分享到：
相关推荐
RxJava&RxAndroid;笔记。RxJava&RxAndroid;笔记 RxJava&RxAndroid;笔记
Rxjava操作符,中文文档,使用案例,方便随时查阅学习. 安卓艺术开发探索读书笔记,EffectiveJava读书笔记. 收集各大神博客以及安卓笔记,安卓面试笔记等方便手机端查看. 更新说明 v2.2.2 1.增加了一些最近在学习比较好...
接下来的时间，我将持续推出Android中Rxjava2.0的一系列文章，包括原理、操作符、应用场景、背压等等，有兴趣可以继续关注Carson_Ho的安卓开发笔记！注：关于Rxjava中的延时创建操作符interval
在Android开发过程中，选择合适的工具能够大大提高开发效率。因此，我们整理了常用的Android开发工具集，包括开发工具、测试工具、性能优化工具等。这些工具都是经过我们精心筛选和测试的，能够帮助开发者们更加高效...
Material Design风格的开源照片笔记。(MVP Dagger2 RxJava Dex分包脚本 Dex异步加载)
在Android开发过程中，选择合适的工具能够大大提高开发效率。因此，我们整理了常用的Android开发工具集，包括开发工具、测试工具、性能优化工具等。这些工具都是经过我们精心筛选和测试的，能够帮助开发者们更加高效...
Java 知识汇总(资源,工具,笔记,源码,文章,文档分类整理).zip ...- 【Rxjava+Retrofit+Okhttp组合使用】在[com.gradle.android.retrofit](/tree/master/src/main/java/com/android/base/net)包名下;
Android学习笔记 对于一个开发者而言，能够胜任系统中任意一个模块的开发是其核心价值的体现。 对于一个架构师而言，掌握各种语言的优势并可以运用到系统中，由此简化系统的开发，是其架构生涯的第一步。 对于一个...
Android Demo, 学习过程中的一些笔记 chapter3 ~ chapter13 为 「Android 群英传」的 Demo ，做了部分修改并在上面增加了一些注释及评论用于备忘 art_Chapter[chapterNumber] 为 「Android 开发艺术探索」的 Demo ，...
该应用程序是使用Google Firebase技术，AAC，Android JetPack，Koin，RxJava和RxAndroid以及一组具有MVVM设计模式的第三方库构建的。 产品特点 丰富的编辑器“ Aztec” 笔记上的CRUD操作 注意时间提醒 注意位置...
学习笔记 设计模式 Android 性能优化 AppApp.svelte 学习博客 Android 开发中的日常积累 目录 Android自定义 Android动画 Android优化 Android值得关注的 ========================================================...
最近在用 MVP + RxJava + Retrofit 写项目，觉得相对于其他的开发框架，这的确是给我们带来了很多方便，但是在网上搜寻相关资料的时候，总是能看到 MVP + RxJava + Retrofit + Dagger 这样的搭配组合，那 Dagger 又...
电影索引 显示来自（TMDb）的电影列表和... 为了构建项目，请在android studio中打开项目，在应用程序的build.gradle文件的第42和48行中，将单词TMDB_API_KEY替换为"\"YOUR_OWN_API_KEY\"" 。 为了获得API_KEY，请在
每日疯狂学习Android的开源项目，基于材料设计+清洁MVP + Weex + Flutter + RxJava2 +改造+ Dagger2 + Glide + OkHttp + MTRVA +炫酷控件+炫酷动画+ greenDAO效果图首页文章：威克斯文章：扑文章：好友列表妹子笔记...
Android UI 生命周期 启动模式 ActivityManagerService PMS Art/Dalvik Http Http WebSocket Udp Kotlin 算法 快速排序 链表 重新排版 RxJava2 Glide GreenDao 其他 反编译 Hook 动态加载 编译构建 设计模式 代理...
笔记 执照 为什么选择MVVM ？ 对于客户端应用程序开发，MVVM比其他MV *模式（例如MVC或MVP）要好。 为什么？ 因为视图模型作为视图数据和行为的额外抽象，因此具有比其他MV *模式更高的抽象级别。 此外，视图模型...
Kotlin-Clean体系结构使用DFM的干净架构Android的ShowCase现代建筑声明式用户界面姓名资料库笔记执照安科不推荐使用Apache许可证V2撰写 Jetpack声明式用户界面Apache许可证V2分裂 Kotlin多平台库（以Android为第一个...
CircularProgressbar项目可让您以最简单的方式在android中创建循环进度条。 用法 为了制作一个圆形的Progressbar，在布局XML中添加CircularProgressbar，并在项目中添加CircularProgressbar库，或者您也可以通过...
leetcode 答案 转载至博客 慕容小匹夫 USEFUL ...：基于RxJava开发的用于帮助在Android 6.0中处理运行时权限检测 ：目前最好用的网络通讯库，应该都用过吧 ：okhttp和retrofit做网络通讯是绝配 ：ORM数
移动安全存储目录项目目的这个项目使我能够研究在移动环境中存储和使用加密数据。 它包含各种简单的演示应用程序，例如笔记，项目和相机应用程序。 该项目利用RxJava允许异步加密和解密数据。 它还利用简单的MVP设计...