使用RxJava实现异步操作(AsyncTask)
常见的异步操作我们可以联想到AsyncTask或者handler,其实google创造出的目的也就是为了让代码更加清晰明了,让代码更加简洁.
而Rx系列的出现也就为了实现代码的逻辑清晰,结构简单问题.在gitHub上的介绍是 a library for composing asynchronous and event-based programs using observable sequences for the Java VM"（一个在 Java VM 上使用可观测的序列来组成异步的、基于事件的程序的库）没错,就是这样,我们可以通过他的操作符,通过他的线程切换,来实现各种Android上遇到的问题.如异步任务,事件总线,数据处理,Retrofit框架等.
提出需求
AsyncTask是异步操作最常见的处理方式,几乎只要涉及到网络请求的,或者耗时操作的,都会使用到AsyncTask,下面看看一个小需求.
- 给一个接口(查询IP)
http://ip.taobao.com/service/getIpInfo.php- 请求参数是
IP地址: ip=192.168.0.1- 功能实现
请求时弹出进度框,请求完成隐藏进度框,并Toast出获取到的数据(需要将数据封装成bean).
常见的异步操作AsyncTask
下面我们使用AsyncTask来实现提出的需求.talk is cheap,请看代码.
//点击发出请求
comTitleStartTv.setOnClickListener(new View.OnClickListener() {
@Override
public void onClick(View v) {
asyncTask();
}
});具体的AsyncTask处理如下.
private void asyncTask() {
new AsyncTask<String, Void, IPInfo>() {
@Override
protected IPInfo doInBackground(String... params) {
// 在后台请求接口,并对请求到的json进行校验,解析成json
String resultJsonStr = OkHttpUtil.get(Constants.ipUrl + getParam(params));
IPInfo infos = null;
if (resultJsonStr != null && !"".equals(resultJsonStr)) {
infos = new Gson().fromJson(resultJsonStr, IPInfo.class);
}
return infos;
}
@Override
protected void onPreExecute() {
super.onPreExecute();
// 请求过程耗时操作,弹出Loading的进度框.
if (dialog == null) {
dialog = ProgressDialog.show(MainActivity.this, "Loading...", "正在加载...", true, false);
}
}
@Override
protected void onPostExecute(IPInfo info) {
// 请求完成,关闭进度框,并Toask请求后的结果数据.
if (dialog != null)
CHelper.disimissDialog(dialog);
if (info != null) {
Toast.makeText(MainActivity.this, "AsyncTask:" + info.toString(), Toast.LENGTH_SHORT).show();
}
}
}.execute("192.168.0.1");
}
protected String getParam(String... p) {
return "?ip=" + p[0];
}下面是AsyncTask的效果. 
RxJava封装的异步操作
下面使用到了RxJava的操作符有线程切换subscribeOn,observeOn,doOnSubscribe,doOnCompleted,doOnError.
- 封装BaseRxTask,主要是抽取公共方法和封装共性的功能.
如:getBaseDilogView主要是封装进度框,处理了io和UI线程.
/**
* @Author: Relice
* BaseRX 异步任务基类
*/
public abstract class BaseRxTask<D, P> {
/**
* 参数数组
*/
protected P[] p;
protected Context mContext;
private boolean needDialog = false;
private ProgressDialog dialog;
/**
* 接口不需要传参数 复写此方法
*/
protected BaseRxTask(Context context, boolean needDialog) {
this.mContext = context;
this.needDialog = needDialog;
}
/**
* 接口需要传参数调 复写此方法
*
* @param context 上下文
* @param needDialog 是否显示进度
* @param p 参数数组
*/
@SafeVarargs
protected BaseRxTask(Context context, boolean needDialog, P... p) {
this.mContext = context;
this.needDialog = needDialog;
this.p = p;
}
protected Observable<D> doInBackgroundObserVable() {
return doInBackground();
}
public Observable<D> execute() {
return getBaseDilogView();
}
/**
* 进度View
*
* @return
*/
private Observable<D> getBaseDilogView() {
Observable<D> tObservable = doInBackgroundObserVable();
//subscribeOn(Schedulers.io())事件所产生的线程,也就是subscribe所发生的线程.
return tObservable.subscribeOn(Schedulers.io())
//doOnSubscribe的作用是如它后面有subscribeOn(),那么它将执行
//离它最近的subscribeOn()所指定的线程,也就是下面的 //observeOn(AndroidSchedulers.mainThread())线程.
.doOnSubscribe(new Action0() {
@Override
public void call() {
if (needDialog && mContext != null) {
dialog = ProgressDialog.show(mContext, "Loading...", "正在加载...", true, false);
}
}
// 注意这里添加了doOnCompleted,doOnError两个预处理的操作,
//所以在后面对Observable的订阅就要用subscribe(new Observer),如果使用subscribe(new Action)会报错.
}).doOnCompleted(new Action0() {
@Override
public void call() {
CHelper.disimissDialog(dialog);
}
}).doOnError(new Action1<Throwable>() {
@Override
public void call(Throwable throwable) {
CHelper.disimissDialog(dialog);
}
})
//为了让进度框可以在UI线程里执行,影响了上面doOnSubscribe的线程
.observeOn(AndroidSchedulers.mainThread());
}
protected abstract Observable<D> doInBackground();
/**
* 接口没参数空实现即可
*
* @param p 参数数组
*/
protected abstract String getParam(P... p);
/**
* @return BaseUrl
*/
protected abstract String getBaseUrl();
}-
实现BaseRxTask
因为在基类对所以的操作都封装好了,所以对BaseRxTask的实现就很简单了,代码中都有注释.public class GetRxIPInfoTask extends BaseRxTask<IPInfo, String> { protected GetRxIPInfoTask(Context context, boolean needDialog) { super(context, needDialog); } public GetRxIPInfoTask(Context context, boolean needDialog, String... p) { super(context, needDialog, p); } @Override public Observable<IPInfo> execute() { return super.execute(); } @Override protected Observable<IPInfo> doInBackground() { return Observable.create(new Observable.OnSubscribe<IPInfo>() { @Override public void call(Subscriber<? super IPInfo> subscriber) { if (!subscriber.isUnsubscribed()) { String resultJsonStr = OkHttpUtil.get(nullToEmpty(getBaseUrl()) + nullToEmpty(getParam(p))); if (resultJsonStr != null && !"".equals(resultJsonStr)) { IPInfo infos = new Gson().fromJson(resultJsonStr, IPInfo.class); subscriber.onNext(infos); subscriber.onCompleted(); } else { subscriber.onCompleted(); } } } }); } private String nullToEmpty(String str) { return (str == null) ? "" : str; } /** * ip * * @param p 参数数组 */ @Override protected String getParam(String... p) { return "?ip=" + p[0]; } /** * @return base url */ @Override protected String getBaseUrl() { return Constants.ipUrl; } } 如何去调用
机智的你们可以看到在基类中的execute()方法封装了 进度框的功能,
public Observable<D> execute() {
return getBaseDilogView();
}具体调用看下面代码
private void rxAsyncTask() {
new GetRxIPInfoTask(MainActivity.this, true, new String[]{"192.168.0.1"})
.execute()
//提前过滤掉不想要的数据
// .filter(new Func1<IPInfo, Boolean>() {
// @Override
// public Boolean call(IPInfo ipInfo) {
// return ipInfo == null;
// }
// })
.subscribe(new Observer<IPInfo>() {
@Override
public void onCompleted() {
System.out.println("onCompleted");
}
@Override
public void onError(Throwable e) {
System.out.println("Throwable:" + e.getMessage());
}
@Override
public void onNext(IPInfo infos) {
System.out.println("onNext:" + infos.toString());
Toast.makeText(MainActivity.this, "RxAsyncTask:" + infos.toString(), Toast.LENGTH_SHORT).show();
}
});
}最后我们看看RxAsyncTask的效果. 