ConcurrencyWithSchedulersDemoFragment
用Schedulers(调度器)实现多任务(并发，Concurrency)的例子
废话不多说我们看一下有关于RxJava的代码：
private Subscription _subscription;
@OnClick(R.id.btn_start_operation) //
public void startLongOperation() {
_progress.setVisibility(View.VISIBLE);
_log("Button Clicked");
_subscription = _getObservable()//
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(_getObserver()); // Observer
}
private Observable<Boolean> _getObservable() {
return Observable.just(true).map(new Func1<Boolean, Boolean>() {
@Override
public Boolean call(Boolean aBoolean) {
_log("Within Observable");
_doSomeLongOperation_thatBlocksCurrentThread();//长时间的操作，阻塞线程
return aBoolean;
}
});
}
private Observer<Boolean> _getObserver() {
return new Observer<Boolean>() {
//....一些操作
};
}这段代码比较简单，只涉及了一个知识点Schedulers调度器
Schedulers
概括：线程调度器，用来达到切换线程的效果。
在Android开发中，由于UI线程是不能够被阻塞的，不然就会产生ANR导致程序崩溃。所以我们经常的处理是，两种方法：
1.新开线程（或者线程池中）处理阻塞的操作：
new Thread(new Runnable() {
@Override
public void run() {
try{
mEntity = getEntity();
mHandler.obtainMessage(1,mEntity).sendToTarget();
}catch (Exception e){
mHandler.sendEmptyMessage(0);
}
}
}).start();大家可以看到，我们这里只是进行了一个简单的IO操作，就陷入了这么谜一般的缩进。要是设计更加复杂的呢？
而且我们还要用Handler处理消息，造成了编码的撕裂感！ 虽然我个人对Handler并不反感，但在编码的时候，它的存在的确让我感到有点惆怅~~
2.回调处理
HttpClient.getInstance().get("http://www.qq.com",new HttpCallback(){
public void onSuccess(String result){
//...成功操作
}
public void onError(Exception e){
//...失败操作
}
});这里我们随便写了一个获取QQ主页的首页数据的操作。 虽然现在看起来还是蛮简单的。 只有一层回调，但是要是我们要针对QQ返回的结果，再进行一次热门词汇的搜索文章，搜索出来的文章我还要获取第五篇文章，然后我再获取第五篇文章的评论呢？
这里就会陷入Callback Hell （回调地狱）！ 维护起来真的是太恶心了！
3.小结一下Schedulers
Schedulers调度器的出现，保持了我们使用RxJava的链式调用，不会出现撕裂感，因为它可以帮我们游刃有余的进行线程切换，而不需要进行Hanlder或者回调！
也正是Schedulers的出现才让我下定决心学习RxJava的初衷！ 这实在的太棒了（当然还有无穷无尽的操作符~~）
结束
1.关于Schedulers的原理
因为能力还没有到这里层度，毕竟RxJava和一般的库不一样，它里面涉及了很多“高端知识”而我也不敢乱说出来误导。 所以呢~这里附上前辈的分析吧（反正我看的是有点一头雾水）：给 Android 开发者的 RxJava 详解 里面有一部分是涉及Schedulers分析的。
我觉得要是看不懂也没关系，随着我们使用的越来越多，没准在哪一天心血来潮想看看源码的时候突然就明白了呢~
在未来的学习，要是我突然明白了，我会尽快写一篇文章和大家讨论的~
2.附录：完整源代码
public class ConcurrencyWithSchedulersDemoFragment
extends BaseFragment {
@Bind(R.id.progress_operation_running)
ProgressBar _progress;
@Bind(R.id.list_threading_log)
ListView _logsList;
private LogAdapter _adapter;
private List<String> _logs;
private Subscription _subscription;
@Override
public void onDestroy() {
super.onDestroy();
RxUtils.unsubscribeIfNotNull(_subscription);
ButterKnife.unbind(this);
}
@Override
public void onActivityCreated(@Nullable Bundle savedInstanceState) {
super.onActivityCreated(savedInstanceState);
_setupLogger();
}
@Override
public View onCreateView(LayoutInflater inflater,
@Nullable ViewGroup container,
@Nullable Bundle savedInstanceState) {
View layout = inflater.inflate(R.layout.fragment_concurrency_schedulers, container, false);
ButterKnife.bind(this, layout);
return layout;
}
@OnClick(R.id.btn_start_operation)
public void startLongOperation() {
_progress.setVisibility(View.VISIBLE);
_log("Button Clicked");
_subscription = _getObservable()//
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(_getObserver()); // Observer
}
private Observable<Boolean> _getObservable() {
return Observable.just(true).map(new Func1<Boolean, Boolean>() {
@Override
public Boolean call(Boolean aBoolean) {
_log("Within Observable");
_doSomeLongOperation_thatBlocksCurrentThread();
return aBoolean;
}
});
}
/** * Observer that handles the result through the 3 important actions: * <p/> * 1. onCompleted * 2. onError * 3. onNext */
private Observer<Boolean> _getObserver() {
return new Observer<Boolean>() {
@Override
public void onCompleted() {
_log("On complete");
_progress.setVisibility(View.INVISIBLE);
}
@Override
public void onError(Throwable e) {
Timber.e(e, "Error in RxJava Demo concurrency");
_log(String.format("Boo! Error %s", e.getMessage()));
_progress.setVisibility(View.INVISIBLE);
}
@Override
public void onNext(Boolean bool) {
_log(String.format("onNext with return value \"%b\"", bool));
}
};
}
// -----------------------------------------------------------------------------------
// Method that help wiring up the example (irrelevant to RxJava)
private void _doSomeLongOperation_thatBlocksCurrentThread() {
_log("performing long operation");
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
Timber.d("Operation was interrupted");
}
}
private void _log(String logMsg) {
if (_isCurrentlyOnMainThread()) {
_logs.add(0, logMsg + " (main thread) ");
_adapter.clear();
_adapter.addAll(_logs);
} else {
_logs.add(0, logMsg + " (NOT main thread) ");
// You can only do below stuff on main thread.
new Handler(Looper.getMainLooper()).post(new Runnable() {
@Override
public void run() {
_adapter.clear();
_adapter.addAll(_logs);
}
});
}
}
private void _setupLogger() {
_logs = new ArrayList<String>();
_adapter = new LogAdapter(getActivity(), new ArrayList<String>());
_logsList.setAdapter(_adapter);
}
private boolean _isCurrentlyOnMainThread() {
return Looper.myLooper() == Looper.getMainLooper();
}
private class LogAdapter
extends ArrayAdapter<String> {
public LogAdapter(Context context, List<String> logs) {
super(context, R.layout.item_log, R.id.item_log, logs);
}
}
}-Hans 2016.3.5 13:36