正文
这可能是最好的RxJava 2.x入门教程系列专栏
文章链接：
这可能是最好的RxJava 2.x 入门教程（完结版）【强力推荐】
这可能是最好的RxJava 2.x 入门教程（一）
这可能是最好的RxJava 2.x 入门教程（二）
这可能是最好的RxJava 2.x 入门教程（三）
这可能是最好的RxJava 2.x 入门教程（四）
这可能是最好的RxJava 2.x 入门教程（五）
GitHub 代码同步更新：https://github.com/nanchen2251/RxJava2Examples
为了满足大家的饥渴难耐，GitHub将同步更新代码，主要包含基本的代码封装，RxJava 2.x所有操作符应用场景介绍和实际应用场景，后期除了RxJava可能还会增添其他东西，总之，GitHub上的Demo专为大家倾心打造。传送门：https://github.com/nanchen2251/RxJava2Examples
一、前言
终于如愿来到让我小伙伴们亢奋的 RxJava 2 使用场景举例了，前面几章中我们讲解完了 RxJava 1.x 到 RxJava 2.x 的异同以及 RxJava 2.x 的各种操作符使用，如有疑问，欢迎点击上方的链接进入你想要的环节。
二、正题
1、简单的网络请求
想必大家都知道，很多时候我们在使用 RxJava 的时候总是和 Retrofit 进行结合使用，而为了方便演示，这里我们就暂且采用 OkHttp3 进行演示，配合 map，doOnNext ，线程切换进行简单的网络请求：
1）通过 Observable.create() 方法，调用 OkHttp 网络请求；
2）通过 map 操作符集合 gson，将 Response 转换为 bean 类；
3）通过 doOnNext() 方法，解析 bean 中的数据，并进行数据库存储等操作；
4) 调度线程，在子线程中进行耗时操作任务，在主线程中更新 UI ；
5) 通过 subscribe()，根据请求成功或者失败来更新 UI 。
Observable.create(new ObservableOnSubscribe<Response>() {
@Override public void subscribe(@NonNull ObservableEmitter<Response> e) throws Exception {
Builder builder = new Builder()
.url("http://api.avatardata.cn/MobilePlace/LookUp?key=ec47b85086be4dc8b5d941f5abd37a4e&mobileNumber=13021671512")
.get();
Request request = builder.build();
Call call = new OkHttpClient().newCall(request);
Response response = call.execute();
e.onNext(response);
}
}).map(new Function<Response, MobileAddress>() {
@Override public MobileAddress apply(@NonNull Response response) throws Exception {
Log.e(TAG, "map 线程:" + Thread.currentThread().getName() + "\n"); if (response.isSuccessful()) {
ResponseBody body = response.body(); if (body != null) {
Log.e(TAG, "map:转换前:" + response.body()); return new Gson().fromJson(body.string(), MobileAddress.class);
}
} return null;
}
}).observeOn(AndroidSchedulers.mainThread())
.doOnNext(new Consumer<MobileAddress>() {
@Override public void accept(@NonNull MobileAddress s) throws Exception {
Log.e(TAG, "doOnNext 线程:" + Thread.currentThread().getName() + "\n");
mRxOperatorsText.append("\ndoOnNext 线程:" + Thread.currentThread().getName() + "\n");
Log.e(TAG, "doOnNext: 保存成功：" + s.toString() + "\n");
mRxOperatorsText.append("doOnNext: 保存成功：" + s.toString() + "\n");
}
}).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<MobileAddress>() {
@Override public void accept(@NonNull MobileAddress data) throws Exception {
Log.e(TAG, "subscribe 线程:" + Thread.currentThread().getName() + "\n");
mRxOperatorsText.append("\nsubscribe 线程:" + Thread.currentThread().getName() + "\n");
Log.e(TAG, "成功:" + data.toString() + "\n");
mRxOperatorsText.append("成功:" + data.toString() + "\n");
}
}, new Consumer<Throwable>() {
@Override public void accept(@NonNull Throwable throwable) throws Exception {
Log.e(TAG, "subscribe 线程:" + Thread.currentThread().getName() + "\n");
mRxOperatorsText.append("\nsubscribe 线程:" + Thread.currentThread().getName() + "\n");
Log.e(TAG, "失败：" + throwable.getMessage() + "\n");
mRxOperatorsText.append("失败：" + throwable.getMessage() + "\n");
}
});
为了方便，我们后面的讲解大部分采用开源的 Rx2AndroidNetworking 来处理，数据来源于天狗网等多个公共API接口。
mRxOperatorsText.append("RxNetworkActivity\n");
Rx2AndroidNetworking.get("http://api.avatardata.cn/MobilePlace/LookUp?key=ec47b85086be4dc8b5d941f5abd37a4e&mobileNumber=13021671512")
.build()
.getObjectObservable(MobileAddress.class)
.observeOn(AndroidSchedulers.mainThread()) // 为doOnNext() 指定在主线程，否则报错
.doOnNext(new Consumer<MobileAddress>() {
@Override public void accept(@NonNull MobileAddress data) throws Exception {
Log.e(TAG, "doOnNext:"+Thread.currentThread().getName()+"\n" );
mRxOperatorsText.append("\ndoOnNext:"+Thread.currentThread().getName()+"\n" );
Log.e(TAG,"doOnNext:"+data.toString()+"\n");
mRxOperatorsText.append("doOnNext:"+data.toString()+"\n");
}
})
.map(new Function<MobileAddress, ResultBean>() {
@Override public ResultBean apply(@NonNull MobileAddress mobileAddress) throws Exception {
Log.e(TAG, "\n" );
mRxOperatorsText.append("\n");
Log.e(TAG, "map:"+Thread.currentThread().getName()+"\n" );
mRxOperatorsText.append("map:"+Thread.currentThread().getName()+"\n" ); return mobileAddress.getResult();
}
})
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<ResultBean>() {
@Override public void accept(@NonNull ResultBean data) throws Exception {
Log.e(TAG, "subscribe 成功:"+Thread.currentThread().getName()+"\n" );
mRxOperatorsText.append("\nsubscribe 成功:"+Thread.currentThread().getName()+"\n" );
Log.e(TAG, "成功:" + data.toString() + "\n");
mRxOperatorsText.append("成功:" + data.toString() + "\n");
}
}, new Consumer<Throwable>() {
@Override public void accept(@NonNull Throwable throwable) throws Exception {
Log.e(TAG, "subscribe 失败:"+Thread.currentThread().getName()+"\n" );
mRxOperatorsText.append("\nsubscribe 失败:"+Thread.currentThread().getName()+"\n" );
Log.e(TAG, "失败："+ throwable.getMessage()+"\n" );
mRxOperatorsText.append("失败："+ throwable.getMessage()+"\n");
}
});
2、先读取缓存，如果缓存没数据再通过网络请求获取数据后更新UI
想必在实际应用中，很多时候（对数据操作不敏感时）都需要我们先读取缓存的数据，如果缓存没有数据，再通过网络请求获取，随后在主线程更新我们的UI。
concat 操作符简直就是为我们这种需求量身定做。
concat 可以做到不交错的发射两个甚至多个 Observable 的发射事件，并且只有前一个 Observable 终止(onComplete) 后才会定义下一个 Observable。
利用这个特性，我们就可以先读取缓存数据，倘若获取到的缓存数据不是我们想要的，再调用 onComplete() 以执行获取网络数据的Observable，如果缓存数据能应我们所需，则直接调用 onNext() ，防止过度的网络请求，浪费用户的流量。
Observable<FoodList> cache = Observable.create(new ObservableOnSubscribe<FoodList>() {
@Override public void subscribe(@NonNull ObservableEmitter<FoodList> e) throws Exception {
Log.e(TAG, "create当前线程:"+Thread.currentThread().getName() );
FoodList data = CacheManager.getInstance().getFoodListData(); // 在操作符 concat 中，只有调用 onComplete 之后才会执行下一个 Observable
if (data != null){ // 如果缓存数据不为空，则直接读取缓存数据，而不读取网络数据
isFromNet = false;
Log.e(TAG, "\nsubscribe: 读取缓存数据:" );
runOnUiThread(new Runnable() {
@Override public void run() {
mRxOperatorsText.append("\nsubscribe: 读取缓存数据:\n");
}
});
e.onNext(data);
}else {
isFromNet = true;
runOnUiThread(new Runnable() {
@Override public void run() {
mRxOperatorsText.append("\nsubscribe: 读取网络数据:\n");
}
});
Log.e(TAG, "\nsubscribe: 读取网络数据:" );
e.onComplete();
}
}
});
Observable<FoodList> network = Rx2AndroidNetworking.get("http://www.tngou.net/api/food/list")
.addQueryParameter("rows",10+"")
.build()
.getObjectObservable(FoodList.class); // 两个 Observable 的泛型应当保持一致
Observable.concat(cache,network)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<FoodList>() {
@Override public void accept(@NonNull FoodList tngouBeen) throws Exception {
Log.e(TAG, "subscribe 成功:"+Thread.currentThread().getName() ); if (isFromNet){
mRxOperatorsText.append("accept : 网络获取数据设置缓存: \n");
Log.e(TAG, "accept : 网络获取数据设置缓存: \n"+tngouBeen.toString作 者： 南 尘 
出 处： http://www.cnblogs.com/liushilin/
关于作者：专注于移动前端的项目开发。如有问题或建议，请多多赐教！欢迎加入Android交流群：118116509
版权声明：本文版权归作者和博客园共有，欢迎转载，但未经作者同意必须保留此段声明，且在文章页面明显位置给出原文链接。
特此声明：所有评论和私信都会在第一时间回复。也欢迎园子的大大们指正错误，共同进步。或者直接私信我
声援博主：如果您觉得文章对您有帮助，可以点击文章下部【推荐】或侧边【关注】。您的鼓励是作者坚持原创和持续写作的最大动力！
http://www.cnblogs.com/liushilin/p/7110489.html
更多精彩分享
- 【知识整理】这可能是最好的RxJava 2.x 教程（完结版） 2017-07-21 16:01:48.16
- 【知识整理】这可能是最好的RxJava 2.x 入门教程（三） 2017-07-20 16:34:55.117
- 【知识整理】这可能是最好的RxJava 2.x 入门教程（二） 2017-07-06 16:30:21.07
- 【知识整理】这可能是最好的RxJava 2.x 入门教程（一） 2017-06-21 11:25:11.537
【知识整理】这可能是最好的RxJava 2.x 入门教程（三）