浅析RxJava和RxAndroid关于线程切换和操作符作用

通过查看大神对RxJava一些解析的博客做了一点对于线程切换的总结

各位大神的链接吧：

如果对RxJava一点都不了解可以去扔物线的《给Android开发者的RxJava详解》查看一下 http://gank.io/post/560e15be2dca930e00da1083

OK 首先定义下subscribeOn(Scheduler)和observeOn(Scheduler)这两个方法的作用，subscribeOn方法是指定事件发出的线程，observeOn方法是指定事件消费的线程。

好了咱边看代码边说， 第一个是写的倒计时，在以前写倒计时的时候都是用的Handler或者AsyncTask,由于我们体现的是线程切换所以我就不用Interval （创建一个定时发射整数序列的Observable）了。

/**
* 是用线程切换实现倒计时
*/
Observable.create(new OnSubscribe<Integer>() {

@Override
public void call(Subscriber<? super Integer> arg0) {
// TODO Auto-generated method stub
int i = 5;
while (i >= 0) {
try {
Thread.sleep(1000);
arg0.onNext(i);
} catch (InterruptedException e) {
e.printStackTrace();
}
i--;
}
arg0.onCompleted();
}
}).subscribeOn(Schedulers.io())// 此方法为上面发出事件设置线程为IO线程
.observeOn(AndroidSchedulers.mainThread())// 为消耗事件设置线程为UI线程
.subscribe(new Subscriber<Integer>() {
@Override
public void onStart() {
// TODO Auto-generated method stub
super.onStart();
tv.setClickable(false);
}

@Override
public void onCompleted() {
tv.setClickable(true);
tv.setText("点击进行倒计时");
}

@Override
public void onError(Throwable arg0) {
}

@Override
public void onNext(Integer arg0) {
tv.setText(arg0 + "秒后可再次点击");
}
});

可以看出subscribeOn(Schedulers.io())影响的是他前面执行所在的线程，而observeOn(AndroidSchedulers.mainThread())影响的是他后面执行所在的线程。虽然代码比Handler和AsyncTask多了但是其逻辑简单了，不是么。先是创建了一个Observable 对象 然后创建一个Subscriber对象。在Observable对象里面调用subscribe方法将Subscriber传进去，在子线程里面用回调的方法传出Subscriber对象进行耗时操作调用Subscriber的onNext方法。因为Subscriber切换的是UI线程所以可以更改UI。

若是上面程序结构看不懂那么请看下面没有简化的代码，若是能看懂请略过这一段。

Observable<Integer> observable = Observable.create(
new OnSubscribe<Integer>() {

@Override
public void call(Subscriber<? super Integer> arg0) {
// TODO Auto-generated method stub
int i = 5;
while (i >= 0) {
try {
Thread.sleep(1000);
arg0.onNext(i);
} catch (InterruptedException e) {
e.printStackTrace();
}
i--;
}
arg0.onCompleted();

}

}).subscribeOn(Schedulers.io());//这设置了发送事件所在的的线程
Subscriber<Integer> subscriber = new Subscriber<Integer>() {
@Override
public void onStart() {
// TODO Auto-generated method stub
super.onStart();
tv.setClickable(false);
}

@Override
public void onNext(Integer arg0) {
tv.setText(arg0 + "秒后可再次点击");
}

@Override
public void onError(Throwable arg0) {
// TODO Auto-generated method stub
}

@Override
public void onCompleted() {
tv.setClickable(true);
tv.setText("点击进行倒计时");
}
};
observable
.observeOn(AndroidSchedulers.mainThread())//这设置了消耗事件所在的线程
.subscribe(subscriber);//对象传进去并调用call回调方法

这段代码就是没有简化的代码。再来一个请求网络的代码。其流程就是在子线程里面请求网络然后在UI线程里面将值设置到组件上。

/**
* 网络请求切换线程
*/
Observable
.just("http://www.baidu.com")
.observeOn(Schedulers.io())//这一句就是开了一个新线程，证明了在他之后执行的都是在子线程中进行的。而我没有用subscribeOn来设置
.map(new Func1<String, String>() {

@Override
public String call(String arg0) {
try {
Thread.sleep(20000); //这一句是证明不是在UI线程做的耗时操作
URL url = new URL(arg0);
HttpURLConnection conn = (HttpURLConnection) url
.openConnection();
conn.setRequestMethod("GET");
conn.setReadTimeout(5000);
int code = conn.getResponseCode();
if (code == 200) {
InputStream in = conn.getInputStream();
ByteArrayOutputStream outStream = new ByteArrayOutputStream();
byte[] data = new byte[1024];
int count = -1;
while ((count = in.read(data, 0, 1024)) != -1)
outStream.write(data, 0, count);
data = null;
return new String(outStream
.toByteArray(), "UTF-8");
}
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
return "请求失败";
}
}).observeOn(AndroidSchedulers.mainThread())//这一句是切换了请求后返回的结果在UI线程中进行的。
.subscribe(new Action1<String>() {

@Override
public void call(String arg0) {
tv.setText(arg0);
}
});

OK，到此线程切换演示完毕。 最后做个总结

/**
* 如果在doOnSubscribe()之后指定了subscribeOn()，它决定了doOnSubscribe()在哪种线程中执行。
* （1）doOnSubscribe()之前的subscribeOn()不会影响它。
* （2）doOnSubscribe()之后的subscribeOn()，且是最近的才会影响它。
* subscribeOn在前面，指定事件发出的线程，observeOn写在后面，指定事件消费的线程
* observeOn能控制下面代码的线程，subscribeOn只能在第一次设置的时候起作用
*/

最后推荐一个 《谁来讲讲Rxjava、rxandroid中的操作符的作用?》提问的答案里面涉及到了RxJava操作符的作用 https://www.zhihu.com/question/32209660

再来个hi大头鬼hi的 《深入浅出RxJava四-在Android中使用响应式编程》 连接 http://blog.csdn.net/lzyzsd/article/details/45033611/