标签：
高能预警：对准备了解RxJava 的程序员一个入口
一直听说Rxjava和RxBinding 在java和android开发中非常的牛逼。主要的优点是响应式编程，逻辑区域化。
今天，我也来说一说关于Rxjava的使用方法。后期会逐步讲到RxBinding在android中的使用和Rxjava同Retrofit结合的例子。
最先在网上看其他人的博客，总感觉讲的不够详细也看的云里雾里。后来观摩的是Rxjava内测程序员“大头鬼”翻译的Rxjava的使用方法博客。看后感觉心中对Rxjava毛瑟顿开。
再后来了解了“扔物线”写的文章，详细的介绍了Rxjava的用法。看完后就对这个框架无法自拔。
这里先贴上他们的博客。不过我还是建议先看完我的博客，再去看“大头鬼”翻译的博客，最后加深了解看“扔物线”的文章。
‘大头鬼Bruce’ RxJava
《给 Android 开发者的 RxJava 详解》 作者：扔物线
切入正题：
基本概念：RxJava 以观察者(Observer)和订阅者(Subscriber)为基础的异步响应方式
有几种方法：
//创建一个Observable对象很简单，直接调用Observable.create即可
Observable<String> mObservable1 = Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("aaa");
subscriber.onCompleted();
subscriber.onError(new Throwable("执行异常信息"));
}
});
//创建Observable对象可以简化成一行代码
Observable<String> mObservable1 = Observable.just("Hello, just world!");注意：订阅Observable有两种方法：
1，Subscribe
2，Observer
如果你只想使用基本功能，选择 Observer 和 Subscriber 是完全一样的
（在 RxJava 的 subscribe 过程中，Observer 也总是会先被转换成一个 Subscriber 再使用）
它们的区别在于：
onStart(): 这是 Subscriber 增加的方法。它会在 subscribe 刚开始，而事件还未发送之前被调用，可以用于做一些准备工作，例如数据的清零或重置。这是一个可选方法，默认情况下它的实现为空。但是它总是在 subscribe 所发生的线程被调用，而不能指定线程。要在指定的线程来做准备工作，可以使用 doOnSubscribe() 方法。具体的可以看扔物线的文章。
unsubscribe(): 这是 Subscriber 所实现的另一个接口 Subscription 的方法，用于取消订阅。在这个方法被调用后，Subscriber 将不再接收事件。一般在这个方法调用前，可以使用 isUnsubscribed() 先判断一下状态。 unsubscribe() 这个方法很重要，因为在 subscribe() 之后， Observable 会持有 Subscriber 的引用，这个引用如果不能及时被释放，将有内存泄露的风险。所以最好在不用的时候解除引用关系，以避免内存泄露的发生。
所以，由此看来，我们还是使用Subscriber比较好（因为Observer还是会被转成Subscriber并且多了几个方法嘛~）
订阅的方式也有几种
//OnSubscriber订阅了mObservable
mObservable.subscribe(new OnSubscriber());//处理Observable对象发出的字符串
private class OnSubscriber extends Subscriber<String> {
@Override
public void onCompleted() {
//完成
}
@Override
public void onError(Throwable e) {
//错误
Log.i("test", e.getMessage());
}
@Override
public void onNext(String s) {
//往下执行的
Log.i("test", s);
}
}另外一种：
//OnSubscriber订阅了mObservable1,只接受onNext方法
mObservable.subscribe(new OnAction1()); //简化Subscriber，不关心OnComplete和OnError，可以使用Action1类。
private class OnAction1 implements Action1<String> {
@Override
public void call(String o) {
Log.i("test", o);
}
}//简化Subscriber，不关心OnNext和OnComplete，可以使用Action1类。
private class OnAction2 implements Action1<Throwable> {
@Override
public void call(Throwable o) {
Log.i("test", o.getMessage());
}
}//简化Subscriber，不关心OnNext和OnError，可以使用Action1类。
private class OnAction3 implements Action0 {
@Override
public void call() {
}
}//一行代码订阅
Subscription subscription = Observable.just(rxJavaBean).subscribe(new OnAction1());
subscription.isUnsubscribed();//检测是否被订阅
subscription.unsubscribe();//退订，将从执行点终止这是Rxjava中最重要的部分，也是内容最多的部分。
开头，后面的代码会省略这个头
//操作符（Operators）
Observable
1）create
.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("aaa");
subscriber.onCompleted();
subscriber.onError(new Throwable("执行异常信息"));
}
});顾名思义创建一个Observable对象。
参数传入一个实例化的OnSubscribe
2）just
.just("Hello, just world!");
//或者
.just(new String[]{"hellp","world","!"});just的作用简化了create的过程。
源码中的解释是：返回一个可观测的,发出一个项目,然后完成。
第二种写法相当于多次执行了create方法中的subscriber.onNext(“aaa”);
3）subscribe
subscribe订阅一个被观察者（Observable）。
订阅后，被观察者返回的结果就会在subscribe中展示。
//操作符（Operators）
Observable
.just("hello")
.subscribe(new Subscriber<String>() {
@Override
public void onCompleted() {
//完成
}
@Override
public void onError(Throwable e) {
//错误
}
@Override
public void onNext(String s) {
//执行
//这里的s就是“hello”
}
});4）map
//操作符（Operators）
Observable
.just("hello")
.map(new Func1<Object, String>() {//在执行操作之前做一些事情
@Override
public String call(Object o) {
return o.toString() + "!!!!";
}
})
.map(new OnMap1("aaaa"))
.map(new OnMap1("aaaa"))
.map(new OnMap1("aaaa"))
.subscribe(new OnAction1());map的作用相当于在create或者just方法后，传入原来的数据，并在原来的数据上做处理后再返回。
那么一旦被订阅（subscribe），打印出来的结果就不再是“hello”了，而是“hello!!!!”。
而且，map可以添加多个，改代码最后结果是“hello！！！！aaaaaaaaaaaa”。
5）from
我们先来定义一个方法query
private Observable<List<String>> query(String text) {
final List<String> lists = new ArrayList<>();
for (int i = 0; i < 10; i++) {
lists.add(i + text);
}
return Observable.create(new Observable.OnSubscribe<List<String>>() {
@Override
public void call(Subscriber<? super List<String>> subscriber) {
subscriber.onNext(lists);
subscriber.onCompleted();
subscriber.onError(new Throwable("无错误信息"));
}
});
}该方法用于模拟返回多个结果
from的作用就相当于把一个集合for循环了。
//更多操作符
Observable<List<String>> observable = query("test");
observable.subscribe(new Action1<List<String>>() {
@Override
public void call(List<String> strings) {
//第一种展示方式
for (String str : strings) {
Log.i("test", str);
}
//第二种展示方式
//Observable.from()方法，它接收一个集合作为输入，然后每次输出一个元素给subscriber
Observable.from(strings).subscribe(new Action1<String>() {
@Override
public void call(String s) {
Log.i("test", s + "_tow");
}
});
}
});6）flatMap
flatMap的作用是输出一个新的Observable，且正是我们在Subscriber想要接收的类型。
//Observable.flatMap()接收一个Observable的输出作为输入，同时输出另外一个Observable
//flatMap输出的新的Observable正是我们在Subscriber想要接收的
observable
.flatMap(new Func1<List<String>, Observable<String>>() {
@Override
public Observable<String> call(List<String> urls) {
return Observable.from(urls);
}
})
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
Log.i("test", s + "_three");
}
});这样一来，和上段用from实现的逻辑是一样的但是代码却清晰了许多。
flatMap的作用就是接收observable输出的结果，然后返回一个observable对象。
7）filter
filter能够过滤集合中不符合条件的项目
observable
.flatMap(new Func1<List<String>, Observable<String>>() {
@Override
public Observable<String> call(List<String> strings) {
return Observable.from(strings);
}
})
.filter(new Func1<String, Boolean>() {//filter能够过滤集合中不符合条件的项目
@Override
public Boolean call(String s) {
return !"0_three".equals(s);
}
})
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
Log.i("test", s);
}
});这样，我们得到的集合就剔除了“0_three”这个字段。返回结果中也不会显示。
8）take
只显示规定条数据
observable
.flatMap(new Func1<List<String>, Observable<String>>() {
@Override
public Observable<String> call(List<String> strings) {
return Observable.from(strings);
}
})
.take(5)//只显示五条数据
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
Log.i("test", s);
}
});这样，不管原来集合中有几条数据，都只返回5条
9）doOnNext
doOnNext()允许我们在每次输出一个元素之前做一些额外的事情
observable
.flatMap(new Func1<List<String>, Observable<String>>() {
@Override
public Observable<String> call(List<String> strings) {
return Observable.from(strings);
}
})
.take(5)//只显示五条数据
.doOnNext(new Action1<String>() {//doOnNext()允许我们在每次输出一个元素之前做一些额外的事情
@Override
public void call(String s) {
//这里把获取到的结果缓存到sd卡
}
})
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
Log.i("test", s);
}
});这样，我们在得到结果之前已经将结果缓存到sd卡里面了
10）subscribeOn
使用subscribeOn()指定观察者代码运行的线程；它把以上的代码放在的非ui线程
observable
.flatMap(new Func1<List<String>, Observable<String>>() {
@Override
public Observable<String> call(List<String> strings) {
return Observable.from(strings);
}
})
.take(5)//只显示五条数据
.doOnNext(new Action1<String>() {//doOnNext()允许我们在每次输出一个元素之前做一些额外的事情
@Override
public void call(String s) {
//这里把获取到的结果缓存到sd卡
}
})
.subscribeOn(Schedulers.newThread())//使用subscribeOn()指定观察者代码运行的线程；它把以上的代码放在的非ui线程
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
Log.i("test", s);
}
});碰巧的是它的名字里包含subscribe，但它管理的却是observable的线程。
11）observeOn
使用observerOn()指定订阅者运行的线程
observable
.flatMap(new Func1<List<String>, Observable<String>>() {
@Override
public Observable<String> call(List<String> strings) {
return Observable.from(strings);
}
})
.take(5)//只显示五条数据
.doOnNext(new Action1<String>() {//doOnNext()允许我们在每次输出一个元素之前做一些额外的事情
@Override
public void call(String s) {
//这里把获取到的结果缓存到sd卡
}
})
.subscribeOn(Schedulers.newThread())//使用subscribeOn()指定观察者代码运行的线程；它把以上的代码放在的非ui线程
.observeOn(AndroidSchedulers.mainThread())//使用observerOn()指定订阅者运行的线程
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
Log.i("test", s);
}
});它的作用则是管理subscribe执行的线程了。
12）线程规划
线程规划：
Schedulers.newThread(): 总是启用新线程，并在新线程执行操作。
Schedulers.io(): I/O 操作（读写文件、读写数据库、网络信息交互等）所使用的 Scheduler。行为模式和 newThread() 差不多，区别在于 io() 的内部实现是是用一个无数量上限的线程池，可以重用空闲的线程，因此多数情况下 io() 比 newThread() 更有效率。不要把计算工作放在 io() 中，可以避免创建不必要的线程。
Schedulers.computation(): 计算所使用的 Scheduler。这个计算指的是 CPU 密集型计算，即不会被 I/O 等操作限制性能的操作，例如图形的计算。这个 Scheduler 使用的固定的线程池，大小为 CPU 核数。不要把 I/O 操作放在 computation() 中，否则 I/O 操作的等待时间会浪费 CPU。
Android 还有一个专用的 AndroidSchedulers.mainThread()，它指定的操作将在 Android 主线程运行。
13）将自定义对象封装成Observable对象
//将你的数据封装成Observable对象
newMethod(UserInfoBean);
//将你的数据封装成Observable对象
private Observable<Object> newMethod(Object slowMethod) {
//选择推迟加载
return Observable.defer((Func0<Observable<Object>>) Observable.just(slowMethod));
}14）RxAndroid
很抱歉RxAndroid已经废了。。。。。。还在活跃开发中，接口变动太大
目前RxAndroid已经被RxBinding所替代。
RxBinding 提供了一套在 Android 平台上的基于 RxJava 的 Binding API。所谓 Binding，就是类似设置 OnClickListener 、设置 TextWatcher 这样的注册绑定对象的 API。
这个后期我也会写出来。
好了，Rxjava的常用方法已经介绍完了。包括创建对象，订阅对象，变换对象，线程控制。
另外，在android中Rxjava配合RxBinding 可以胜任android平台下的一些操作。这也是非常方便的。
而Rxjava也被Retrofit内置了。只要添加依赖库，它们两个就可以配合使用了。
compile ‘io.reactivex:rxjava:1.0.14‘
compile ‘io.reactivex:rxandroid:1.0.1‘
compile ‘com.squareup.retrofit:adapter-rxjava:2.0.0-beta2‘
compile ‘com.squareup.retrofit:retrofit:2.0.0-beta2‘
compile ‘com.squareup.retrofit:converter-gson:2.0.0-beta2‘这些依赖库分别是：Rxjava、RxAndroid、retrofit适配Rxjava、retrofit2.0、retrofit转换json数据。
当然还有RxBinding。
RxJava——响应式和区域化的优秀框架（java&android）
标签：
原文地址：http://blog.csdn.net/yehui928186846/article/details/51484198
