RxJava教程(一):响应式编程介绍

简介

响应式编程是一种以异步数据流为核心的编程方式。这里的数据一般是一些事件,而流则是在时间序列上的一系列的事件。任何东西都可以转化为数据流,如变量、用户输入事件、数据结构等。

我们可以很灵活地操纵数据流,如可以将两个甚至多个数据流融合成一个数据流,可以从数据流中过滤出感兴趣的事件,还可以将数据流中的事件转化为其他新的事件。

数据流中的事件通常可以分成三种类型:普通事件、错误事件和结束事件。以用户的键盘输入事件为例,当用户依次敲击“A”、“B”、“C”键的时候,就会产生三个输入事件,计算机接收到这些事件并对其做出响应一一将字母“ A”、“B ”、“ C”显示在显示器上。当用户敲击回车键时,可以将其作为一个结束事件来表示数据流的结束,即用户输入结束。而在输入过程中发生的任何错误都可以作为数据流中的错误事件。

继续阅读“RxJava教程(一):响应式编程介绍”
0

使用Retrofit + RxJava 实现优雅的处理服务器返回异常、错误

异常&错误

实际开发经常有这种情况,比如登录请求,接口返回的

信息包括请求返回的状态:失败还是成功,错误码,User对象等等。如果网络等原因引起的登录失败可以归结为异常,如果是用户信息输入错误导致的登录失败算是错误。

假如服务器返回的是统一数据格式:

/**

* 标准数据格式

* @param <T>

*/

public class Response<T> {

public int state;

public String message;

public T data;

}

网络异常导致的登录失败,在使用Retrofit+RxJava请求时都会直接调用subscribe的onError事件;

密码错误导致的登录失败,在使用Retrofit+RxJava请求时都会调用subscribe的onNext事件;

无论是异常还是错误,都要在subscribe里面处理异常信息,如下代码:

APIWrapper.getInstance().login(“username”, “password”)

.subscribe(new Observer<Response<User>>() {

@Override

public void onCompleted() {

}

@Override

public void onError(Throwable e) {

}

@Override

public void onNext(Response<User> data) {

if(data.state == 1001){

//…..

}else if(data.state == 1002){

}

}

});

现在我希望在发生任何错误的情况下,都会调用onError事件,并且由model来处理错误信息。那么,此时我们就应该有一个ExceptionEngine来处理事件流中的错误信息了。

在工作流中处理异常

在正常情况下,我们获取网络数据的流程通常如下:

请求接口->解析数据->更新UI

整个数据请求过程都是发生在Rx中的工作流之中。当有异常产生的时候,我们要尽量不在ui层里面进行判断,换句话说,我们没有必要去告诉ui层具体的错误信息,只需要让他弹出一个信息(Toast或者Dialog)展示我们给它的信息就行。

请求接口和数据解析都可能出错,所以在这两层进行错误处理。为了更好的解耦,我们通过拦截器拦截错误,然后根据错误类型分发信息。

拦截器

数据解析层的拦截器

这个拦截器主要是为了获取具体的错误信息,分发给上层的UI,给用户以提示,增强用户体验。

public Observable<Weather> getWeather(String cityName){

return weatherService.getWeather(cityName)

//拦截服务器返回的错误

.map(new ServerResponseFunc<Weather>())

//HttpResultFunc()为拦截onError事件的拦截器,后面会讲到,这里先忽略

.onErrorResumeNext(new HttpResponseFunc<Weather>());

}

//拦截固定格式的公共数据类型Response<T>,判断里面的状态码

private class ServerResponseFunc<T> implements Func1<Response<T>, T> {

@Override

public T call(Response<T> reponse) {

//对返回码进行判断,如果不是0,则证明服务器端返回错误信息了,便根据跟服务器约定好的错误码去解析异常

if (reponse.state != 0) {

//如果服务器端有错误信息返回,那么抛出异常,让下面的方法去捕获异常做统一处理

throw new ServerException(reponse.state,reponse.message);

}

//服务器请求数据成功,返回里面的数据实体

return reponse.data;

}

}

所以整个逻辑是这样的:

所以在前三步的过程中,只要发生异常(服务器返回的错误也抛出了)都会抛出,这时候就触发了RxJava的OnError事件。

处理onError事件的拦截器

这个拦截器主要是将异常信息转化为用户”能看懂”的友好提示。

private class HttpResponseFunc<T> implements Func1<Throwable, Observable<T>> {

@Override

public Observable<T> call(Throwable throwable) {

//ExceptionEngine为处理异常的驱动器

return Observable.error(ExceptionEngine.handleException(throwable));

}

}

两个拦截器以前使用,代码如下:

public Observable<Weather> getWeather(String cityName){

return weatherService.getWeather(cityName)

//拦截服务器返回的错误

.map(new ServerResponseFunc<Weather>())

//HttpResponseFunc()为拦截onError事件的拦截器

.onErrorResumeNext(new HttpResponseFunc<Weather>());

}

调用:

APIWrapper.getInstance().getWeather(“北京”)

.subscribe(new SampleProgressObserver<Weather>(MainActivity.this) {

@Override

public void onNext(WeatherBean weatherBean) {

tv.setText(weatherBean.toString());

}

});

相关类:

public class RxSubscriber<T> extends ErrorSubscriber<T> {

@Override

public void onStart() {

super.onStart();

DialogHelper.showProgressDlg(context, “正在加载数据”);

}

@Override

public void onCompleted() {

DialogHelper.stopProgressDlg();

}

@Override

protected void onError(ApiException ex) {

DialogHelper.stopProgressDlg();

Toast.makeText(context, ex.message, Toast.LENGTH_SHORT).show();

}

@Override

public void onNext(T t) {

}

}

public abstract class ErrorSubscriber<T> extends Observer<T> {

@Override

public void onError(Throwable e) {

if(e instanceof ApiException){

onError((ApiException)e);

}else{

onError(new ApiException(e,123));

}

}

/**

* 错误回调

*/

protected abstract void onError(ApiException ex);

}

处理异常的驱动器

package com.v791202;

import android.net.ParseException;

import com.google.gson.JsonParseException;

import org.json.JSONException;

import java.net.ConnectException;

import retrofit2.adapter.rxjava.HttpException;

/**

* Created by 791202.com

*/

public class ExceptionEngine {

//对应HTTP的状态码

private static final int UNAUTHORIZED = 401;

private static final int FORBIDDEN = 403;

private static final int NOT_FOUND = 404;

private static final int REQUEST_TIMEOUT = 408;

private static final int INTERNAL_SERVER_ERROR = 500;

private static final int BAD_GATEWAY = 502;

private static final int SERVICE_UNAVAILABLE = 503;

private static final int GATEWAY_TIMEOUT = 504;

public static ApiException handleException(Throwable e){

ApiException ex;

if (e instanceof HttpException){ //HTTP错误

HttpException httpException = (HttpException) e;

ex = new ApiException(e, ERROR.HTTP_ERROR);

switch(httpException.code()){

case UNAUTHORIZED:

case FORBIDDEN:

case NOT_FOUND:

case REQUEST_TIMEOUT:

case GATEWAY_TIMEOUT:

case INTERNAL_SERVER_ERROR:

case BAD_GATEWAY:

case SERVICE_UNAVAILABLE:

default:

ex.message = “网络错误”; //均视为网络错误

break;

}

return ex;

} else if (e instanceof ServerException){ //服务器返回的错误

ServerException resultException = (ServerException) e;

ex = new ApiException(resultException, resultException.code);

ex.message = resultException.message;

return ex;

} else if (e instanceof JsonParseException

|| e instanceof JSONException

|| e instanceof ParseException){

ex = new ApiException(e, ERROR.PARSE_ERROR);

ex.message = “解析错误”; //均视为解析错误

return ex;

}else if(e instanceof ConnectException){

ex = new ApiException(e, ERROR.NETWORD_ERROR);

ex.message = “连接失败”; //均视为网络错误

return ex;

}else {

ex = new ApiException(e, ERROR.UNKNOWN);

ex.message = “未知错误”; //未知错误

return ex;

}

}

}

/**

* 约定异常

*/

public class ERROR {

/**

* 未知错误

*/

public static final int UNKNOWN = 1000;

/**

* 解析错误

*/

public static final int PARSE_ERROR = 1001;

/**

* 网络错误

*/

public static final int NETWORD_ERROR = 1002;

/**

* 协议出错

*/

public static final int HTTP_ERROR = 1003;

}

public class ApiException extends Exception {

public int code;

public String message;

public ApiException(Throwable throwable, int code) {

super(throwable);

this.code = code;

}

}

public class ServerException extends RuntimeException {

public int code;

public String message;

}

DialogHelper.Java

public class DialogHelper {

/**

* 通用Dialog

*

*/

// 因为本类不是activity所以通过继承接口的方法获取到点击的事件

public interface OnOkClickListener {

abstract void onOkClick();

}

/**

* Listener

*/

public interface OnCancelClickListener {

abstract void onCancelClick();

}

private static AlertDialog mDialog;

public static void showDialog(Context context, String title, String content, final OnOkClickListener listenerYes,

final OnCancelClickListener listenerNo) {

showDialog(context, context.getString(android.R.string.ok), context.getString(android.R.string.cancel), title, content, listenerYes, listenerNo);

}

public static void showDialog(Context context, String ok, String cancel, String title, String content, final OnOkClickListener listenerYes,

final OnCancelClickListener listenerNo) {

AlertDialog.Builder builder = new AlertDialog.Builder(context);

builder.setMessage(content);

// 设置title

builder.setTitle(title);

// 设置确定按钮,固定用法声明一个按钮用这个setPositiveButton

builder.setPositiveButton(ok,

new DialogInterface.OnClickListener() {

public void onClick(DialogInterface dialog, int which) {

// 如果确定被电击

if (listenerYes != null) {

listenerYes.onOkClick();

}

mDialog = null;

}

});

// 设置取消按钮,固定用法声明第二个按钮要用setNegativeButton

builder.setNegativeButton(cancel,

new DialogInterface.OnClickListener() {

public void onClick(DialogInterface dialog, int which) {

// 如果取消被点击

if (listenerNo != null) {

listenerNo.onCancelClick();

}

mDialog = null;

}

});

// 控制这个dialog可不可以按返回键,true为可以,false为不可以

builder.setCancelable(false);

// 显示dialog

mDialog = builder.create();

if (!mDialog.isShowing())

mDialog.show();

}

public static void showDialog(Context context, int ok, int cancel, int title, int content, final OnOkClickListener listenerYes,

final OnCancelClickListener listenerNo) {

showDialog(context, context.getString(ok), context.getString(cancel), context.getString(title), context.getString(content), listenerYes, listenerNo);

}

static ProgressDialog progressDlg = null;

/**

* 启动进度条

*

* @param strMessage 进度条显示的信息

* @param // 当前的activity

*/

public static void showProgressDlg(Context ctx, String strMessage) {

if (null == progressDlg) {

if (ctx == null) return;

progressDlg = new ProgressDialog(ctx);

//设置进度条样式

progressDlg.setProgressStyle(ProgressDialog.STYLE_SPINNER);

//提示的消息

progressDlg.setMessage(strMessage);

progressDlg.setIndeterminate(false);

progressDlg.setCancelable(true);

progressDlg.show();

}

}

public static void showProgressDlg(Context ctx) {

showProgressDlg(ctx, “”);

}

/**

* 结束进度条

*/

public static void stopProgressDlg() {

if (null != progressDlg && progressDlg.isShowing()) {

progressDlg.dismiss();

progressDlg = null;

}

if (null != dialog && dialog.isShowing()) {

dialog.dismiss();

dialog = null;

}

}

private static Dialog dialog;

public static void showDialogForLoading(Context context, String msg, boolean cancelable) {

if (null == dialog) {

if (null == context) return;

View view = LayoutInflater.from(context).inflate(R.layout.layout_loading_dialog, null);

TextView loadingText = (TextView)view.findViewById(R.id.loading_tip_text);

loadingText.setText(msg);

dialog = new Dialog(context, R.style.loading_dialog_style);

dialog.setCancelable(cancelable);

dialog.setCanceledOnTouchOutside(cancelable);

dialog.setContentView(vwww.791202.comiew, new LinearLayout.LayoutParams(LinearLayout.LayoutParams.MATCH_PARENT, LinearLayout.LayoutParams.MATCH_PARENT));

Activity activity = (Activity) context;

if (activity.isFinishing()) return;

dialog.show();

}

}

}

可能本博客也不是最好的解决方案,如果有更好的想法,我愿与你互相交流!

分享: Retrofit+RxJava错误预处理

看到bobo_wang的文章,不仅感觉有受益匪浅,这里做下介绍。

首先定义如下Transformer转换器。

public static <T> Observable.Transformer<Response<T>, T> sTransformer() {

return responseObservable -> responseObservable.map(tResponse -> {

if (!tResponse.success) throw new RuntimeException(tResponse.code);

return tResponse.dwww.791202.comata;

}).onErrorResumeNext(new HttpResponseFunc<>());

}

public static <T> Observable.Transformer<T, T> switchSchedulers() {

return observable -> observable.subscribeOn(Schedulers.io())

.observeOn(AndroidSchedulers.mainThread());

}

private static class HttpResponseFunc<T> implements Func1<Throwable, Observable<T>> {

@Override public Observable<T> call(Throwable throwable) {

//ExceptionEngine为处理异常的驱动器

return Observable.error(new Throwable(throwable));

}

}

调用:

public void login(View v){

apiservice.login(name,pwd)

.compose(Transformers.sTransformer())

.compose(Transformers.switchSchedulers())

.subscribe(subscriber);

}

private Subscriber<UserModel> subscriber = new Subscriber<UserModel>() {

@Override public void onCompleted() {

// do onCompleted

}

@Override public void onError(Throwable e) {

// do on success != true;

// do on http error

// do on other error

}

@Override public void onNext(UserModel model) {

// parse data

}

};

接口:

@FormUrlEncoded @POST(“interface?login”)

Observable<Response<UserModel>> login(@Field(“name”) String name,@Field(“pwd”) String pwd);

最后再来点干货。

Transformer 和 Func 处理的区别

如上的处理,定义了 一个 sTransformer 和一个 HttpResponseFunc,

从中可以明显感觉的到sTransformer其实也是可以用Func1来定义的,

public void login(View v){

apiservice.login(name,pwd)

.compose(Transformers.switchSchedulers())

.map(new TransFuc<UserModel>())

.onErrorReturn(new HttpResponseFunc<>())

.subscribe(subscriber);

}

public static class TransFuc<T> implements Func1<Response<T>, T> {

@Override public T call(Response<T> tResponse) {

if (!tResponse.success) throw new RuntimeException(tResponse.code);

return tResponse.data;

}

}

Transformer作用于整个流,Func1是一个操作符,作用于数据项。

不规范数据的处理

有时候服务器返回的数据并不是十分规范的,比如

正常返回是这样的

{

“success”: true, // 是否成功

“status”: “1”, // 状态码

“data”: {

// 内容

}

}

错误时时这样的

{

“success”: false, // 是否成功

“status”: “0”, // 状态码

“data”: “371” //错误码

}

这时候如果我么用泛型处理

public class Response<T> {

public boolean success;

public String status;

public T data;

}

针对这种数据,我们的泛型该怎么写成了问题,错误的时候是String,正确的时候是Bean?

如果我们直接写成JavaBean,那么我们会得到一个错误,LinkedTreeMap cannot be cast to xxx

类型转换错误.这时候只能将泛型写成String来处理了,使用如下Transformer

public static Observable.Transformer<String, UserModel> trans() {

return stringObservable -> stringObservable.map(s -> {

Response parseJson = GsonUtil.parseJson(s, Response.class);

if (null == parseJson) {

throw new RuntimeException(“null == parseJson”);

}

if (PatternsUtil.isNum(parseJson.data.toString())) {

throw new RuntimeException(parseJson.data.toString());

}

return GsonUtil.parseJson(s, UserModel.class);

}).onErrorResumeNext(new HttpResponseFunc<>());

}

使用就变成了如下这样

public void login(View v){

apiservice.login(name,pwd)

.compose(Transformers.switchSchedulers())

.compose(Transformers.trans())

.subscribe(subscriber);

}

封装的越来越简介了,用到实际项目吧!

0