使用RxJava實現(xiàn)異步操作(AsyncTask)

常見的異步操作我們可以聯(lián)想到AsyncTask或者handler,其實google創(chuàng)造出的目的也就是為了讓代碼更加清晰明了,讓代碼更加簡潔.
而Rx系列的出現(xiàn)也就為了實現(xiàn)代碼的邏輯清晰,結(jié)構(gòu)簡單問題.在gitHub上的介紹是 a library for composing asynchronous and event-based programs using observable sequences for the Java VM"（一個在 Java VM 上使用可觀測的序列來組成異步的、基于事件的程序的庫）沒錯,就是這樣,我們可以通過他的操作符,通過他的線程切換,來實現(xiàn)各種Android上遇到的問題.如異步任務(wù),事件總線,數(shù)據(jù)處理,Retrofit框架等.

提出需求

AsyncTask是異步操作最常見的處理方式,幾乎只要涉及到網(wǎng)絡(luò)請求的,或者耗時操作的,都會使用到AsyncTask,下面看看一個小需求.

給一個接口(查詢IP)
http://ip.taobao.com/service/getIpInfo.php

請求參數(shù)是
IP地址: ip=192.168.0.1

功能實現(xiàn)
請求時彈出進度框,請求完成隱藏進度框,并Toast出獲取到的數(shù)據(jù)(需要將數(shù)據(jù)封裝成bean).

常見的異步操作AsyncTask

下面我們使用AsyncTask來實現(xiàn)提出的需求.talk is cheap,請看代碼.

//點擊發(fā)出請求 comTitleStartTv.setOnClickListener(new View.OnClickListener() {@Overridepublic void onClick(View v) {asyncTask();}});

具體的AsyncTask處理如下.

private void asyncTask() {new AsyncTask<String, Void, IPInfo>() {@Overrideprotected IPInfo doInBackground(String... params) {// 在后臺請求接口,并對請求到的json進行校驗,解析成jsonString resultJsonStr = OkHttpUtil.get(Constants.ipUrl + getParam(params));IPInfo infos = null;if (resultJsonStr != null && !"".equals(resultJsonStr)) {infos = new Gson().fromJson(resultJsonStr, IPInfo.class);}return infos;}@Overrideprotected void onPreExecute() {super.onPreExecute();// 請求過程耗時操作,彈出Loading的進度框.if (dialog == null) {dialog = ProgressDialog.show(MainActivity.this, "Loading...", "正在加載...", true, false);}}@Overrideprotected void onPostExecute(IPInfo info) {// 請求完成,關(guān)閉進度框,并Toask請求后的結(jié)果數(shù)據(jù).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 異步任務(wù)基類*/ public abstract class BaseRxTask<D, P> {/*** 參數(shù)數(shù)組*/protected P[] p;protected Context mContext;private boolean needDialog = false;private ProgressDialog dialog;/*** 接口不需要傳參數(shù) 復寫此方法*/protected BaseRxTask(Context context, boolean needDialog) {this.mContext = context;this.needDialog = needDialog;}/*** 接口需要傳參數(shù)調(diào) 復寫此方法** @param context 上下文* @param needDialog 是否顯示進度* @param p 參數(shù)數(shù)組*/@SafeVarargsprotected 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())事件所產(chǎn)生的線程,也就是subscribe所發(fā)生的線程.return tObservable.subscribeOn(Schedulers.io())//doOnSubscribe的作用是如它后面有subscribeOn(),那么它將執(zhí)行 //離它最近的subscribeOn()所指定的線程,也就是下面的 //observeOn(AndroidSchedulers.mainThread())線程..doOnSubscribe(new Action0() {@Overridepublic 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() {@Overridepublic void call() {CHelper.disimissDialog(dialog);}}).doOnError(new Action1<Throwable>() {@Overridepublic void call(Throwable throwable) {CHelper.disimissDialog(dialog);}}) //為了讓進度框可以在UI線程里執(zhí)行,影響了上面doOnSubscribe的線程.observeOn(AndroidSchedulers.mainThread());}protected abstract Observable<D> doInBackground();/*** 接口沒參數(shù)空實現(xiàn)即可** @param p 參數(shù)數(shù)組*/protected abstract String getParam(P... p);/*** @return BaseUrl*/protected abstract String getBaseUrl(); }

• 實現(xiàn)BaseRxTask
  因為在基類對所以的操作都封裝好了,所以對BaseRxTask的實現(xiàn)就很簡單了,代碼中都有注釋.

  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>() {@Overridepublic 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 參數(shù)數(shù)組*/ @Override protected String getParam(String... p) {return "?ip=" + p[0]; }/*** @return base url*/ @Override protected String getBaseUrl() {return Constants.ipUrl; } }

• 如何去調(diào)用
  機智的你們可以看到在基類中的execute()方法封裝了 進度框的功能,

public Observable<D> execute() {return getBaseDilogView();}

具體調(diào)用看下面代碼

private void rxAsyncTask() {new GetRxIPInfoTask(MainActivity.this, true, new String[]{"192.168.0.1"}).execute()//提前過濾掉不想要的數(shù)據(jù) // .filter(new Func1<IPInfo, Boolean>() { // @Override // public Boolean call(IPInfo ipInfo) { // return ipInfo == null; // } // }).subscribe(new Observer<IPInfo>() {@Overridepublic void onCompleted() {System.out.println("onCompleted");}@Overridepublic void onError(Throwable e) {System.out.println("Throwable:" + e.getMessage());}@Overridepublic void onNext(IPInfo infos) {System.out.println("onNext:" + infos.toString());Toast.makeText(MainActivity.this, "RxAsyncTask:" + infos.toString(), Toast.LENGTH_SHORT).show();}});}

最后我們看看RxAsyncTask的效果.

總結(jié)

以上是生活随笔為你收集整理的你知道RxJava也可以实现AsyncTask吗?的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。