RxJava 之 Android使用初体验

什么是RxJava

RxJava 就是异步

RxJava 的异步实现，是通过一种扩展的观察者模式来实现的。

一个响应式编程框架

通过之前对RxJava的了解，我们已经掌握了RxJava的一些基础使用方法，下面我们结合一个简单的Demo，看看如何在Android 开发中使用RxJava。

RxAndroid 初体验

需求

这里我们的需求很简单：

点击按钮，执行一个网络请求，将返回的json信息解析，实现UI 更新

这也是我们做APP最常用的套路。首先看一下，我们需要 实现的效果。



看到网上关于RxJava的网络请求的内容，都会提及Retrofit的使用，然而由于Retrofit使用了注解相关的内容，代码看起来会有点不好理解，这里我们就从最基础的网络请求出发，一步一步理解一下RxAndroid的使用。

RxAndroid+OkHttp+Gson 实现数据更新

OkHttp

OKHttp 大家应该都了解，OkHttp 是基于http协议封装的一套请求客户端。具体的细节就不展开说了，这里直接使用。

我们创建OKHttpClient 和 Request

client = new OkHttpClient();
request = new Request.Builder()
.url(BaseUrl)
.build();

创建Observeable （OKHttp 同步执行）

private Observable<Response> HttpService() {
Observable myObserve;

myObserve = Observable.create(new Observable.OnSubscribe<Response>() {
@Override
public void call(Subscriber<? super Response> subscriber) {
Response response = null;
try {
response = client.newCall(request).execute();
subscriber.onNext(response);
subscriber.onCompleted();

} catch (IOException e) {
subscriber.onError(e);
}
}

});

这里我们通过oncreate 操作符创建了一个Observeable，在其call 方法中执行了OkHttp的一个同步get请求，

网络请求正常时，将请求响应Resoponse通过onNext方法返回，同时执行onCompleted方法。

网络请求异常时，将异常信息通过onError方法返回。

订阅者执行subscribe

private void getData() {
HttpService()
.subscribe(new Subscriber<Response>() {
@Override
public void onCompleted() {
Log.e("onCompleted", "onCompleted");
}

@Override
public void onError(Throwable e) {
Log.e("onError", e.getMessage());
}

@Override
public void onNext(Response response) {
if (response.isSuccessful()) {
try {
String json = response.body().string();
Log.e("onNext", json);
setView(json);
} catch (IOException e) {
e.printStackTrace();
}
}

}
});
}

通过之前的学习，我们知道在订阅者（Subscribe）实现订阅（subscribe）操作的同时，Observeable就会开始发送事件，在我们当前的例子里，就是开始执行oncall方法，通过OKHttp的网络请求，将请求到的信息Response通过onNext返回，而在订阅者的onNext 方法里，我们处理Response信息，实现UI更新。

指定正确的执行线程

你应该已经注意到，上面一个简单的流程里，我们进行了网络请求、UI更新等操作，而这些操作必然是在不同的线程中完成，这就要求我们整个操作流程必须实现正确的线程切换，RxJava 固然强大，但也无法实现线程的智能切换 ，必须由我们去指定合适的线程。所以我们上面的代码是有问题的，需要进行如下修改：

private void getData() {
HttpService()
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Subscriber<Response>() {
onCompleted();
......unchange
}

这里我们的修改很简单，两行关键代码：

.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())

这两行代码指定了，我们的网络请求在io线程执行，不会阻塞主线程；UI更新确保在主线程中执行。

总结来说，这里我们使用到了RxJava中的调度器，就是subscribeOn 指定了Observeable 执行的线程；而observeOn指定了Subscribe（订阅者）执行的线程。

这样我们非常简单实现了对线程的控制。

以正确的方式更新UI

好了，我们现在可以放心的执行getData方法了 ，在其onNext 方法，我们接收Response中的数据：

这里分享一下，我在使用这个OKHttp 的Response时遇到一个坑：

Log.e("onNext", response.body().string());
String json = response.body().string();
Log.e("onNext", json);

执行上面的三行代码，你会发现，第一次打印的日志是有数据的，而第二次打印的日志数据却是null，第一次看到这个情况的时候，真的是让我惊呆了，后来看了body 方法的API才明白：

/**
* Never {@code null}, must be closed after consumption, can be consumed only once.
*/
public ResponseBody body() {
return body;
}

原来这个boby方法只能被执行一次。这个真是有点坑啊！！！！每次获取数据都有打印日志的习惯，真不知道这么做的意义是什么。

好了，我们继续，数据已经能正确接收了，接下来就是解析数据并更新UI 了。

public void setView(String json) {
Gson gson = new Gson();
DoubanBean douban = new DoubanBean();
douban = gson.fromJson(json, DoubanBean.class);
//
Glide.with(mContext).load(douban.getIcon()).into(pic);
title.setText(douban.getTitle());
id.setText(douban.getId());

}

这里我们用Gson 解析返回的json 格式数据，实现View 内容更新。

这里我们调用getData方法，结果发现页面没有任何变化，看Logcat日志输出：

E/onError: Permission denied (missing INTERNET permission?)

原来是我们忘记在AndroidManifest 文件中声明网络请求的权限了，这里执行了onError 方法，返回了异常信息。

我们加上INTERNET 相关的权限之后，运行程序，发现可以正常执行了。

创建Observeable （OKHttp 异步执行）

之前我们创建Observeable的方式是通过OKHttp的同步get请求方式，如果用异步请求方式怎么做呢？

// OkHttp 异步执行
myObserve=Observable.create(new Observable.OnSubscribe<Response>() {
@Override
public void call(final Subscriber<? super Response> subscriber) {
client.newCall(request).enqueue(new Callback() {
@Override
public void onFailure(Call call, IOException e) {
subscriber.onError(e);
}

@Override
public void onResponse(Call call, Response response) throws IOException {
subscriber.onNext(response);
subscriber.onCompleted();
}
});
}
});

看以看到，和同步请求方式不同的是，我们将onNext ,onCompleted 和 onError这些回调方法放在了OKHttp自己的回调里进行执行。这样做的唯一优点可能就是我们在执行订阅的时候不用指定Observeable（网络请求）执行的线程了，因为他本身就是异步的，就是这里：

private void getData() {
HttpService()
//okHttp 异步执行时，无需指定网络请求的线程
//               .subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.......
}

当然，OkHttp终究只是对网络请求做了一次封装的库，和最基础的HttpClient, HttpUrlConnection的功能是一样的，不像Volley、android-async-http这些库；无论同步执行还是异步执行，其请求结果始终是在子线程，所以我们的Subscribe对于UI的更新还是需要指定其为mainThread 的。

这里可以看出，使用异步的OKHttp反而有点多余了。

后话

这里我们通过 RxAndroid+OkHttp+Gson 的方式，以一种最费劲的方式实现了一个简单的需求（这个Demo真的很烂），只是为了阐述RxJava 的编程思想在Android中使用的体验。这里的需求，通过AsnycTask 或者OkHttp + Handler 的方式实现是非常简单的，就不再详述了。

RxJava 的作用就是实现异步，如果我们原本的操作本来就是异步，为了使用RxJava而硬套进去是不合理的，反而显得有点不伦不类，这里我们从Observeable的创建使用OkHttp 的同步&异步请求就可以看出，使用异步请求并没有多大益处，反而丢掉了RxJava可以指定线程的优点。

RxJava的使用并不能使我们的程序运行更高效，或者可以实现了别的框架实现不了的功能。只是对我们编写代码的方式提供了一种更好的思路。

好了，RxJava 基础到这里就结束了。文中所讲的Demo在GitHub中也有，感兴趣的可以点这里。