这篇博文酝酿好久了，今天终于下定决心开始写！RxAndroid和Retrofit都算是当下非常流行的Android开发框架，这两个框架光是单独使用就已经爽歪歪了，那么将RxAndroid和Retrofit结合起来，又会有怎样的效果呢？
鉴于很多小伙伴可能还没用过RxJava或者RxAndroid，所以我打算先来介绍一下RxJava和RxAndroid，然后再来介绍RxAndroid和Retrofit组合拳！！
本文主要包括以下三方面内容：
1.RxJava简介
2.RxJava在Android中的使用
3.RxJava配合Retrofit的使用
OK，废话不多说，那就开始吧！
1.RxJava简介
RxJava作为Android开发者的新宠已经有很长一段时间了，用过RxJava的人，都觉得这个东西简单好用，没用过的小伙伴第一次看到RxJava的代码时可能会觉得这个东西非常繁琐，那么今天，我想通过几个简单的非HelloWorld的案例来让大家彻底理解RxJava的使用。
先来看看RxJava的GitHub地址：
https://github.com/ReactiveX/RxJava
将RxJava引入到我们的项目中：
compile 'io.reactivex:rxandroid:1.1.0'
compile 'io.reactivex:rxjava:1.1.0'
那么学习RxJava，先得弄清楚什么是RxJava，关于RxJava的介绍，GitHub上有一句话，
a library for composing asynchronous and event-based programs by using observable sequences.
翻译成中文就是：
RxJava是一个基于可观测序列组成的异步的、基于事件的库。通俗一点说就是RxJava它是一个异步库，这个异步库可以让我们用非常简洁的代码来处理复杂数据流或者事件。OK，那么这是对RxJava的一个基本介绍，接下来我们再来看看RxJava中两个最最基础的概念，一个是Observable，还有一个是Observer，其中Observable我们称之为被观察者，Observer称之为观察者，Observable用户发送消息，而Observer用于消费消息，在实际开发中，我们更多的是选择Observer的一个子类Subscriber来消费消息。在消息发送的过程中，Observable可以发送任意数量任意类型的消息（甚至一个将一个触摸事件当作一个消息发出），当Observable所发送的消息被成功处理或者消息出错时，一个流程结束。Observable会用它的每一个Subscriber（观察者）的onNext方法进行消息处理，在消息成功处理后以onComplete()方法结束流程，如果消息在处理的过程中出现了任何异常，则以onError()方法结束流程。比如下面几行代码：
Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("哈哈哈哈");
subscriber.onNext("lalalala");
subscriber.onCompleted();
}
})
.subscribe(new Observer<String>() {
@Override
public void onCompleted() {
Log.d("google_lenve_fb", "onCompleted: onCompleted()");
}
@Override
public void onError(Throwable e) {
Log.d("google_lenve_fb", "onError:onError() ");
}
@Override
public void onNext(String s) {
Log.d("google_lenve_fb", "onNext: " + s);
}
});
通过调用Observable的create方法来创建一个消息源，在它的onCall方法调用next方法来发送两条消息，当两条消息发送完成之后，调用onComplete方法表示消息发送完毕！subscribe表示订阅一条消息，在订阅的时候我们可以传入一个Observer对象，也可以传入一个Subscriber对象，这两个对象中的方法都是一样的，在onNext方法中处理消息，当消息处理完成之后会自动的调用onComplete方法，如果消息处理过程中出错，则会调用onError方法，上面方法打印的日志如下：
onNext: 哈哈哈哈 onNext: lalalala onCompleted: onCompleted()
如果我在onNext方法执行一行 1/0 ，onNext方法改成下面的样子：
@Override
public void onNext(String s) {
Log.d("google_lenve_fb", "onNext: " + s);
int i = 1 / 0;
}
这个时候再运行，系统打印的日志如下：
onNext: 哈哈哈哈 onError:onError()
当第一条消息打印出来之后，执行1/0时抛异常，直接调用了onError方法，第二条消息将不再处理。OK，上面的代码是一个小小的案例，RxJava中还有许多好玩的操作符，我们接下来一个一个来看。
在实际开发中，我们的数据来源可能是多种多样的，为了简化数据操作，Observable类为我们提供了许多现成的方法，这些方法都能够极大的简化我们对数据的操作，一个一个来看吧。
1.1 from函数
Observale中的from函数接受一个数组，这个方法返回一个按数组元素的顺序来发射这些数据的Observale，看看下面一行代码：
Observable.from(new String[]{"竹外桃花三两枝", "春江水暖鸭先知"})
.subscribe(new Subscriber<String>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(String s) {
Log.d("google_lenve_fb", "onNext: " + s);
}
});
这里的观察者我使用了Subscriber，它是Observer的一个实现类。这里打印的结果如下：
D/google_lenve_fb: onNext: 竹外桃花三两枝 D/google_lenve_fb: onNext: 春江水暖鸭先知
按顺序将from函数中数组的值一个一个打印出来了。OK，接着往下看：
1.2 just函数
just函数它接受最多10个参数，返回一个按参数顺序发射这些数据的Observable，代码如下：
Observable.just("Hello", "World", "Hello", "RxJava!","塞外秋来风景异","衡阳雁去无留意"
,"四面边声连角起","千嶂里","长烟落日孤城闭","浊酒一杯家万里")
.subscribe(new Subscriber<String>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(String s) {
Log.d("google_lenve_fb", "onNext: " + s);
}
});
打印结果如下：
D/google_lenve_fb: onNext: Hello D/google_lenve_fb: onNext: World D/google_lenve_fb: onNext: Hello D/google_lenve_fb: onNext: RxJava! D/google_lenve_fb: onNext: 塞外秋来风景异 D/google_lenve_fb: onNext: 衡阳雁去无留意 D/google_lenve_fb: onNext: 四面边声连角起 D/google_lenve_fb: onNext: 千嶂里 D/google_lenve_fb: onNext: 长烟落日孤城闭 D/google_lenve_fb: onNext: 浊酒一杯家万里
1.3map函数
map函数可以对Observable创建的原始数据进行二次加工，然后再被观察者获取。比如下面一段代码，我给原始数据的每一项都追加一个字符串，然后返回：
Observable.from(new String[]{"醉里挑灯看剑","梦回吹角连营"})
.map(new Func1<String, String>() {
@Override
public String call(String s) {
return s + "---辛弃疾";
}
})
.subscribe(new Observer<String>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(String s) {
Log.d("google_lenve_fb", "onNext: " + s);
}
});
打印结果如下：
D/google_lenve_fb: onNext: 醉里挑灯看剑---辛弃疾 D/google_lenve_fb: onNext: 梦回吹角连营---辛弃疾
RxJava的使用都是链式编程，使用map函数时数据来源可以各种各样。
1.4flatMap函数
flatMap函数接受一个Observable函数作为输入函数，然后在这个输入的基础上再创建一个新的Observable进行输出，比如下面一段代码：
Observable.just("落霞与孤鹜齐飞", "秋水共长天一色")
.flatMap(new Func1<String, Observable<String>>() {
@Override
public Observable<String> call(String s) {
return Observable.from(new String[]{s + "冯唐易老", s + "李广难封"});
}
}).subscribe(new Subscriber<String>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(String s) {
Log.d("google_lenve_fb", "onNext: " + s);
}
});
打印结果：
D/google_lenve_fb: onNext: 落霞与孤鹜齐飞冯唐易老 D/google_lenve_fb: onNext: 落霞与孤鹜齐飞李广难封 D/google_lenve_fb: onNext: 秋水共长天一色冯唐易老 D/google_lenve_fb: onNext: 秋水共长天一色李广难封
1.5 scan函数
scan函数是一个累加器函数，对于Observable发射的每项数据进行累加，并将累加的结果返回，如下：
Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
.scan(new Func2<Integer, Integer, Integer>() {
@Override
public Integer call(Integer integer, Integer integer2) {
return integer + integer2;
}
})
.subscribe(new Observer<Integer>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Integer integer) {
Log.d("google_lenve_fb", "onNext: " + integer);
}
});
Observable.from(new String[]{"明", "月", "别", "枝", "惊", "鹊"})
.scan(new Func2<String, String, String>() {
@Override
public String call(String s, String s2) {
return s + s2;
}
})
.subscribe(new Subscriber<String>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(String s) {
Log.d("google_lenve", "onNext: " + s);
}
});
打印结果如下：
D/google_lenve_fb: onNext: 1 D/google_lenve_fb: onNext: 3 D/google_lenve_fb: onNext: 6 D/google_lenve_fb: onNext: 10 D/google_lenve_fb: onNext: 15 D/google_lenve_fb: onNext: 21 D/google_lenve_fb: onNext: 28 D/google_lenve_fb: onNext: 36 D/google_lenve_fb: onNext: 45 D/google_lenve_fb: onNext: 55 D/google_lenve: onNext: 明 D/google_lenve: onNext: 明月 D/google_lenve: onNext: 明月别 D/google_lenve: onNext: 明月别枝 D/google_lenve: onNext: 明月别枝惊 D/google_lenve: onNext: 明月别枝惊鹊
1.6elementAt函数
elementAt函数表示获取数据源中的第N项数据输出，如下：
Observable.just("清", "风", "半", "夜", "鸣", "蝉")
.elementAt(4)
.subscribe(new Subscriber<String>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(String s) {
Log.d("google_lenve", "onNext: "+s);
}
});
打印结果如下：
D/google_lenve: onNext: 鸣
1.7merge函数
merge函数可以用来合并多个Observable数据源，然后将合并后的数据在一起输出，如下：
Observable<String> observable1 = Observable.just("十里楼台倚翠薇", "百花深处杜鹃啼");
Observable<String> observable2 = Observable.just("殷勤自与行人语","不似流莺取次飞");
Observable.merge(observable1, observable2)
.subscribe(new Subscriber<String>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(String s) {
Log.d("google_lenve", "onNext: "+s);
}
});
打印结果如下：
D/google_lenve: onNext: 十里楼台倚翠薇 D/google_lenve: onNext: 百花深处杜鹃啼 D/google_lenve: onNext: 殷勤自与行人语 D/google_lenve: onNext: 不似流莺取次飞
1.8 zip函数
zip函数用来合并多个Observable的数据源，但是与merge不同的是，zip函数中可以对数据源进行二次操作，而不是简单的合并，代码如下：
Observable<String> observable1 = Observable.just("十里楼台倚翠薇", "百花深处杜鹃啼");
Observable<String> observable2 = Observable.just("问君能有几多愁","恰似一江春水向东流");
Observable.zip(observable1, observable2, new Func2<String, String, Object>() {
@Override
public Object call(String s, String s2) {
return s+s2;
}
})
.subscribe(new Subscriber<Object>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Object o) {
Log.d("google_lenve", "onNext: "+o.toString());
}
});
打印日志如下：
D/google_lenve: onNext: 十里楼台倚翠薇问君能有几多愁 D/google_lenve: onNext: 百花深处杜鹃啼恰似一江春水向东流
1.9其他的过滤函数
Observable函数中还有其他一些好用的过滤函数，我就不再一一演示了，小伙伴们看下面的代码自行试验即可：
private void m7() {
Observable.just(10, 12, 13, 14, 15, 16, 16, 16, 17, 18)
//在数据序列的开头插入一条指定的项1
.startWith(2)
.filter(new Func1<Integer, Boolean>() {
@Override
public Boolean call(Integer integer) {
return integer > 0;
}
})
//只发射前N个元素
.take(2)
//只发射最后N个元素
.takeLast(2)
//只发射第一个元素
.first()
//只发射最后一个元素
.last()
//跳过前两个
.skip(2)
//跳过最后两个
.skipLast(2)
//数据过滤，过滤掉重复数据
.distinct()
.subscribe(new Observer<Integer>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Integer integer) {
Log.d("google_lenve_fb", "onNext: " + integer);
}
});
1.10 subscribeOn函数和observeOn函数
这两个函数在我们Android开发中还是非常有用的，其中subscribeOn表示指定被观察者执行的线程，而observeOn则表示观察者执行的线程，这个解决网络访问还是非常方便的，如下：
Observable<String> observable1 = Observable.just("十里楼台倚翠薇", "百花深处杜鹃啼");
Observable<String> observable2 = Observable.just("问君能有几多愁","恰似一江春水向东流");
Observable.zip(observable1, observable2, new Func2<String, String, Object>() {
@Override
public Object call(String s, String s2) {
Log.d("google_lenve", "call: "+Thread.currentThread().getName());
return s+s2;
}
})
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Subscriber<Object>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Object o) {
Log.d("google_lenve", "onNext: "+Thread.currentThread().getName()+"--------"+o.toString());
}
});
打印日志如下：
D/google_lenve: call: RxCachedThreadScheduler-1 D/google_lenve: call: RxCachedThreadScheduler-1 D/google_lenve: onNext: main--------十里楼台倚翠薇问君能有几多愁 D/google_lenve: onNext: main--------百花深处杜鹃啼恰似一江春水向东流
2.RxJava在Android中的使用
RxJava在安卓开发中不仅能够处理简单的数据流，最好玩的地方莫过于网络访问，我举一个例子（其他的数据流处理这里就不再介绍了），比如我有一个按钮，点击该按钮加载一张图片出来，代码如下：
Observable.create(new Observable.OnSubscribe<Bitmap>() {
@Override
public void call(Subscriber<? super Bitmap> subscriber) {
subscriber.onNext(getBitmap());
}
})
//设置数据加载在子线程进行
.subscribeOn(Schedulers.io())
//设置图片加载在主线程进行
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Subscriber<Bitmap>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Bitmap bitmap) {
iv.setImageBitmap(bitmap);
}
});
getBitmap方法如下：
private Bitmap getBitmap() {
HttpURLConnection con;
try {
URL url = new URL("http://img3.cache.netease.com/photo/0008/2016-08-10/BU4A81A72ODN0008.550x.0.jpg");
con = (HttpURLConnection) url.openConnection();
con.setConnectTimeout(5 * 1000);
con.connect();
if (con.getResponseCode() == 200) {
return BitmapFactory.decodeStream(con.getInputStream());
}
} catch (MalformedURLException e) {
e.printStackTrace();
} catch (IOException e) {
e.printStackTrace();
}
return null;
}
有没有感觉爽歪歪呢？从此彻底告别Handler，告别new Thread。。。。。
3.RxJava配合Retrofit的使用
最后我们再来看看当下最流行的两个框架的组合使用！如果你还不了解Retrofit的使用，参考我之前的博客一个App带你学会Retrofit2.0，麻麻再也不用担心我的网络请求了！
将RxJava和Retrofit结合的时候，首先先修改gradle文件，先把我的gradle文件拿出来给大家瞧瞧：
compile 'com.squareup.retrofit2:retrofit:2.1.0'
compile 'com.squareup.retrofit2:adapter-rxjava:2.1.0'
compile 'com.squareup.retrofit2:converter-gson:2.1.0'
compile 'io.reactivex:rxandroid:1.1.0'
compile 'io.reactivex:rxjava:1.1.0'
APIService中的定义如下：
public interface APIService {
@GET("api/lore/classify")
Observable<ClassfyBean> getClassfyBean();
}
创建Retrofit同时创建APIService的实例，代码如下：
Retrofit retrofit = new Retrofit.Builder()
.baseUrl("http://www.tngou.net")
.addConverterFactory(GsonConverterFactory.create())
.addCallAdapterFactory(RxJavaCallAdapterFactory.create())
.build();
APIService apiService = retrofit.create(APIService.class);
注意，这里和我们在一般场景下使用Retrofit有一个区别，那就是多了addCallAdapterFactory这个方法，这里我们添加的RxJava的适配工厂。然后我们就可以调用APIService中的方法来访问网络数据了，如下：
observable
//在子线程访问数据
.subscribeOn(Schedulers.io())
//在主线程显示数据
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Subscriber<ClassfyBean>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable throwable) {
}
@Override
public void onNext(ClassfyBean classfyBean) {
StringBuffer sb = new StringBuffer();
for (ClassfyBean.TngouBean tngouBean : classfyBean.getTngou()) {
sb.append(tngouBean.getName() + "\n");
}
tv.setText(sb.toString());
}
});
OK，这才是网络访问和数据处理最佳拍档。
今天先这样，后面再找时间详细介绍这两个结合的条条框框。。。。
以上。
另外推荐两篇大神文章：
http://wuxiaolong.me/2016/01/18/rxjava/
http://gank.io/post/560e15be2dca930e00da1083