RxJava2修炼之路（三）

作者: luweicheng24 | 来源:发表于2017-09-06 18:51 被阅读0次

RxJava2修炼之路（三）—— 操作符

在上一节中了解了在主线程和子线程中如何进行切换和线程的调度，能够很好的处理在不同的线程中处理不同的任务，本节说一下RxJava中的操作符：

map操作符是RxJava中一个最基础也是最重要的操作符，主要是用来对被观察者发送的数据在半路进行数据类型一对一转换之后进行继续发送给观察者：
举例：需要将一个List中的Integer类型的数据加10后转换成字符转继续发送给观察者：

  public void demoMap() {
        List<Integer> mList = new ArrayList<>();
        for (int i = 0; i < 3; i++) {
            mList.add((int) (Math.random() * 10));
        }
        Observable.fromIterable(mList)
                .map(new Function<Integer, String>() {
                    @Override
                    public String apply(Integer integer) throws Exception {
                        Log.d(TAG, "apply: "+integer);
                        return (integer + 10) + "";
                    }
                }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                String result = s;
                Log.d(TAG, "accept: " + s);
            }
        });
    }

打印结果：

09-06 16:26:09.417 12892-12892/ruanrong.com.rxjava2demo D/test: apply: 3
09-06 16:26:09.417 12892-12892/ruanrong.com.rxjava2demo D/test: accept: 13
09-06 16:26:09.417 12892-12892/ruanrong.com.rxjava2demo D/test: apply: 6
09-06 16:26:09.417 12892-12892/ruanrong.com.rxjava2demo D/test: accept: 16
09-06 16:26:09.417 12892-12892/ruanrong.com.rxjava2demo D/test: apply: 4
09-06 16:26:09.417 12892-12892/ruanrong.com.rxjava2demo D/test: accept: 14

flatMap

flatMap是一个将一个事件发送的Observable转换成多个Observable之后这多个Observable封装成一个新的Observable继续发送：
举例：发送三个Integer数字，中间将每个数字转换成一个三个事件之后迭代发送给订阅者：

public void flatMapDemo(){
         Observable.just(1,2,3,4,5)
                 .flatMap(new Function<Integer, ObservableSource<String>>() {
             @Override
             public ObservableSource<String> apply(Integer integer) throws Exception {
                 List<String> mList = new ArrayList<String>();
                 for (int i = 0; i < 2; i++) {
                     mList.add("i am number is "+integer);
                 }
                 return Observable.fromIterable(mList).delay(5, TimeUnit.SECONDS);
             }
         }).subscribe(new Observer<String>() {
             @Override
             public void onSubscribe(Disposable d) {
                 Log.d(TAG, "onSubscribe: ");
             }

             @Override
             public void onNext(String value) {
                 Log.d(TAG, "onNext: "+value);
             }

             @Override
             public void onError(Throwable e) {

             }

             @Override
             public void onComplete() {
                 Log.d(TAG, "onComplete: ");
             }
         });
    }

打印结果：

09-06 17:15:00.147 13200-13200/ruanrong.com.rxjava2demo D/test: onSubscribe: 
09-06 17:15:05.157 13200-13236/ruanrong.com.rxjava2demo D/test: onNext: i am number is 1
09-06 17:15:05.157 13200-13236/ruanrong.com.rxjava2demo D/test: onNext: i am number is 1
09-06 17:15:05.157 13200-13237/ruanrong.com.rxjava2demo D/test: onNext: i am number is 2
09-06 17:15:05.157 13200-13237/ruanrong.com.rxjava2demo D/test: onNext: i am number is 2
09-06 17:15:05.157 13200-13238/ruanrong.com.rxjava2demo D/test: onNext: i am number is 3
09-06 17:15:05.157 13200-13239/ruanrong.com.rxjava2demo D/test: onNext: i am number is 4
09-06 17:15:05.157 13200-13238/ruanrong.com.rxjava2demo D/test: onNext: i am number is 3
09-06 17:15:05.157 13200-13240/ruanrong.com.rxjava2demo D/test: onNext: i am number is 5
09-06 17:15:05.157 13200-13239/ruanrong.com.rxjava2demo D/test: onNext: i am number is 4
09-06 17:15:05.157 13200-13240/ruanrong.com.rxjava2demo D/test: onNext: i am number is 5
09-06 17:15:05.157 13200-13240/ruanrong.com.rxjava2demo D/test: onComplete:

进过flatMap过的新的Observable会没有顺序将变化后的Observable发送出来，上面添加的延时是为了验证发送顺序的无序性，既然flatMap是无序的，自然有有序的吧？没错，就是 concatMap顾名思义就是有序的Map，那到底是不是呢？是骡子是马。拉出来溜溜就知道了.

concatMap

和flatMap同样的问题，来看一下结果如何：

public void concatMapDemo(){
         Observable.just(1,2,3,4,5)
                 .concatMap(new Function<Integer, ObservableSource<String>>() {
             @Override
             public ObservableSource<String> apply(Integer integer) throws Exception {
                 List<String> mList = new ArrayList<String>();
                 for (int i = 0; i < 2; i++) {
                     mList.add("i am number is "+integer);
                 }
                 return Observable.fromIterable(mList).delay(5, TimeUnit.SECONDS);
             }
         }).subscribe(new Observer<String>() {
             @Override
             public void onSubscribe(Disposable d) {
                 Log.d(TAG, "onSubscribe: ");
             }

             @Override
             public void onNext(String value) {
                 Log.d(TAG, "onNext: "+value);
             }

             @Override
             public void onError(Throwable e) {

             }

             @Override
             public void onComplete() {
                 Log.d(TAG, "onComplete: ");
             }
         });
    }

打印结果：

09-06 17:23:51.597 21178-21216/ruanrong.com.rxjava2demo D/test: onNext: i am number is 1
09-06 17:23:51.597 21178-21216/ruanrong.com.rxjava2demo D/test: onNext: i am number is 1
09-06 17:23:56.597 21178-21297/ruanrong.com.rxjava2demo D/test: onNext: i am number is 2
09-06 17:23:56.597 21178-21297/ruanrong.com.rxjava2demo D/test: onNext: i am number is 2
09-06 17:24:01.597 21178-21372/ruanrong.com.rxjava2demo D/test: onNext: i am number is 3
09-06 17:24:01.597 21178-21372/ruanrong.com.rxjava2demo D/test: onNext: i am number is 3
09-06 17:24:06.597 21178-21453/ruanrong.com.rxjava2demo D/test: onNext: i am number is 4
09-06 17:24:06.597 21178-21453/ruanrong.com.rxjava2demo D/test: onNext: i am number is 4
09-06 17:24:11.597 21178-21531/ruanrong.com.rxjava2demo D/test: onNext: i am number is 5
09-06 17:24:11.597 21178-21531/ruanrong.com.rxjava2demo D/test: onNext: i am number is 5
09-06 17:24:11.597 21178-21531/ruanrong.com.rxjava2demo D/test: onComplete:

没错，果然是有序的，这里如果是flatMap是将所有事件都存在一个自带的缸里面，然后等所有都转换完成后一股脑全部发送，但是concatMap确实每一个原始事件转换成新的Observable后就发送了，在上面的例子中就是每次发两个相同的数据。那这个操作符到底有什么作用呢？比如在日常开发者，有个登录注册的功能，注册完成直接登录，首先利用Eclipse和tomcat搭建一个简单的登录注册：
首先来申明一下登录注册都需要两个字段:

String name;
String password;

登录成功返回字段：

{"state":1,"message":"登录成功","userId":"0931","token":"7a638f1hfh974434"}

登录失败返回字段：

{"state":0,"message":"用户名或者密码不正确"}

注册返回的和登录返回的基本差不多，只是注册成功后没有返回userid和token这两个字段，由于只是简单的模仿登录注册，后台的返回Json是拼接而成，后台代码如下：

/**
 * 
 * 作者： 卢卫成
 * 时间： 2017年9月6日
 * 功能描述：登录
 *
 */
public class LoginServlet  extends HttpServlet{
    @Override
    protected void doGet(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {
        this.doPost(req, resp);
        System.out.println("doget");
        
    }
   @Override
protected void doPost(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {
    StringBuilder builder = new StringBuilder("{");
    builder.append("\"state\":1,");
    builder.append("\"message\":\"登录成功\",");
    builder.append("\"userId\":\"0931\",");
    builder.append("\"token\":\"7a638f1hfh974434\"}");
    StringBuilder fail = new StringBuilder("{");
    fail.append("\"state\":0,");
    resp.setContentType("text/plain");
    resp.setCharacterEncoding("UTF-8");
    String name = (String)req.getParameter("name");
    String psw  = (String)req.getParameter("password");
    PrintWriter out = resp.getWriter();
    if(name!=null&&psw!=null){
        if(name.equals("kobe")&&psw.equals("123")){
            out.write(builder.toString());
        }
        else{
            fail.append("\"message\":\"用户名或者密码不正确\"}");
            out.write(fail.toString());
        }
    }else{
        fail.append("\"message\":\"用户名或者密码为空\"}");
        out.write(fail.toString());
    }
    out.flush();
    out.close();
}
}


/**
 * 
 * 作者： 卢卫成
 * 时间： 2017年9月6日
 * 功能描述： 注册
 *
 */
public class RegistServlet  extends HttpServlet{
    @Override
    protected void doGet(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {
        this.doPost(req, resp);
        System.out.println("doget");
        
    }
   @Override
protected void doPost(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {
    StringBuilder success = new StringBuilder("{");
    success.append("\"state\":1,");
    success.append("\"message\":\"注册成功\"}");
    StringBuilder fail = new StringBuilder("{");
    fail.append("\"state\":0,");
    resp.setContentType("text/plain");
    resp.setCharacterEncoding("UTF-8");
    String name = (String)req.getParameter("name");
    String psw  = (String)req.getParameter("password");
    PrintWriter out = resp.getWriter();
    if(name!=null&&psw!=null){
        if(name.length()>0&&name.length()<16&&psw.length()==3){
            out.write(success.toString());
        }
        else{
            fail.append("\"message\":\"注册失败,账号不符合格式\"}");
            out.write(fail.toString());
        }
    }else{
        fail.append("\"message\":\"注册失败,账号为空\"}");
        out.write(fail.toString());
    }
    out.flush();
    out.close();
}
}

在Android端，网络请求采用的Retrofit2来进行的网络请求,Retrofit的实例以及api如下

// api接口

public interface ApiService {

    @FormUrlEncoded
    @POST("login")
    Observable<LoginModel> login(@FieldMap HashMap<String,String> map);

    @FormUrlEncoded
    @POST("regist")
    Observable<RegistModel> regist(@FieldMap HashMap<String,String> map);
}

**
 * Author   : luweicheng on 2017/9/3 0003 13:38
 * E-mail   ：1769005961@qq.com
 * GitHub   : https://github.com/luweicheng24
 * funcation: Retrofit 网络管理类
 */

public class RetrofitManager {
    private static final String BASE_URL = "http://192.168.81.2:8080/RetrofitServer/";
    private static ApiService apiService;
    private RetrofitManager() {
    }
    public static ApiService getApi() {
        if (apiService == null) {
            synchronized (RetrofitManager.class) {
                if (apiService == null) {
                    /**
                     *  log 的拦截器
                     */
                    HttpLoggingInterceptor interceptor = new HttpLoggingInterceptor()
                            .setLevel(HttpLoggingInterceptor.Level.BODY);
                    /**
                     * OkHttpClient 客户端
                     */
                    OkHttpClient client = new OkHttpClient
                            .Builder()
                            .addInterceptor(interceptor)//日志过滤器
                            .connectTimeout(8 * 1000, TimeUnit.SECONDS) // 连接事件
                            .retryOnConnectionFailure(true) // 是否重新连接
                            .build();
                    /**
                     *  Retrofit 对象
                     */
                    Retrofit retrofit = new Retrofit
                            .Builder()
                            .baseUrl(BASE_URL)
                            .client(client)
                            .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
                            .addConverterFactory(GsonConverterFactory.create())
                            .build();
                    apiService = retrofit.create(ApiService.class);
                }

            }
        }
        return apiService;
    }
}

接下来就是利用flatMap来进行注册完成后自动登录功能：

public void regist(View view){
    String acc = name.getText().toString();
    String password = psw.getText().toString();
    if(TextUtils.isEmpty(acc)||TextUtils.isEmpty(password)){
        return;
    }else{
        final HashMap<String ,String > map = new HashMap<>();
        map.put("name",acc);
        map.put("password",password);
        RetrofitManager.getApi()
                .regist(map)
                .subscribeOn(Schedulers.io()) // 注册放io线程
                .observeOn(AndroidSchedulers.mainThread()) // 注册完成切换到主线程
                .doOnNext(new Consumer<RegistModel>() {
                    @Override
                    public void accept(RegistModel registModel) throws Exception {
                        // 根据注册返回信息本地处理
                        Log.d(TAG, "accept: "+registModel.toString());
                    }
                })
                .observeOn(Schedulers.io())//登录切换到io线程
                .flatMap(new Function<RegistModel, ObservableSource<LoginModel>>() {
                    @Override
                    public ObservableSource<LoginModel> apply(RegistModel registModel) throws Exception {
                         Observable observable= RetrofitManager.getApi().login(map);
                        return  observable;
                    }
                })
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<LoginModel>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                    }
                    @Override
                    public void onNext(LoginModel value) {
                  //登录成功
                        Log.d(TAG, "onNext: 登录成功");
                    }
                    @Override
                    public void onError(Throwable e) {
                        Log.d(TAG, "onError: "+e.toString());
                    }

                    @Override
                    public void onComplete() {
                        Log.d(TAG, "onComplete: ");
                    }
                });
    }

}

打印日志：

09-06 18:48:14.977 13171-13171/ruanrong.com.retrofit2demo D/tag: accept: RegistModel{state=1, message='注册成功'}
09-06 18:48:15.007 13171-13171/ruanrong.com.retrofit2demo D/tag: onNext: 登录成功
09-06 18:48:15.007 13171-13171/ruanrong.com.retrofit2demo D/tag: onComplete:

结果也显示注册完成后登录成功，利用flatMap完成注册登录功能，好了，下班了。

RxJava2修炼之路（三）

RxJava2修炼之路（三）—— 操作符

map

flatMap

concatMap

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读