美文网首页
RxJava的操作符

RxJava的操作符

作者: kdong | 来源:发表于2018-11-28 16:08 被阅读0次

    一、Map操作符

    map是RxJava中最简单的一个变换操作符了, 它的作用就是对上游发送的每一个事件应用一个函数, 使得每一个事件都按照指定的函数去变化。
    这边直接以代码的形式去展示下map操作符的使用:

    Observable.create(new ObservableOnSubscribe<Integer>() {
    
               @Override
               public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
    
                   emitter.onNext(10);
                   emitter.onNext(20);
                   emitter.onNext(30);
    
               }
           }).map(new Function<Integer, String>() {
    
               @Override
               public String apply(Integer integer) throws Exception {
                   return "接收到:"+integer;
               }
           }).subscribe(new Consumer<String>() {
               @Override
               public void accept(String s) throws Exception {
                   Log.e(TAG, s);
               }
           });
        }
    

    看下打印结果:

    11-28 15:17:58.233 14116-14116/com.example.qidong.testrxjava E/MainActivity: 接收到:10
    11-28 15:17:58.234 14116-14116/com.example.qidong.testrxjava E/MainActivity: 接收到:20
    11-28 15:17:58.234 14116-14116/com.example.qidong.testrxjava E/MainActivity: 接收到:30
    

    被观察者发送的是Integer类型的数据,经过map变换之后,观察者接收到的是String类型的数据。通过Map, 可以将上游发来的事件转换为任意的类型, 可以是一个Object, 也可以是一个集合,功能是非常的强大。

    二、flatMap操作符
    FlatMap将一个发送事件的上游Observable变换为多个发送事件的Observables,然后将它们发射的事件合并后放进一个单独的Observable里。

    Observable.create(new ObservableOnSubscribe<Integer>() {
    
                @Override
                public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                    emitter.onNext(1);
                    emitter.onNext(2);
                    emitter.onNext(3);
    
                }
            }).flatMap(new Function<Integer, ObservableSource<String>>() {
                @Override
                public ObservableSource<String> apply(Integer integer) throws Exception {
                    List<String> list=new ArrayList<>();
                    for(int i=0;i<3;i++)
                    {
                        list.add(integer+":"+i);
                    }
                    //延迟5秒
                    return Observable.fromIterable(list).delay(10, TimeUnit.MILLISECONDS);
                }
            }).subscribe(new Consumer<String>() {
                @Override
                public void accept(String s) throws Exception {
                    Log.e(TAG,s);
                }
            });
    

    如代码所示, 我们在flatMap中将上游发来的每个事件转换为一个新的发送三个String事件的水管, 为了看到flatMap结果是无序的,所以加了10毫秒的延时, 来看看运行结果吧:

    11-28 15:38:40.751 16137-16208/com.example.qidong.testrxjava E/MainActivity: 2:0
    11-28 15:38:40.756 16137-16208/com.example.qidong.testrxjava E/MainActivity: 2:1
    11-28 15:38:40.756 16137-16208/com.example.qidong.testrxjava E/MainActivity: 2:2
    11-28 15:38:40.759 16137-16209/com.example.qidong.testrxjava E/MainActivity: 3:0
    11-28 15:38:40.764 16137-16209/com.example.qidong.testrxjava E/MainActivity: 3:1
    11-28 15:38:40.765 16137-16209/com.example.qidong.testrxjava E/MainActivity: 3:2
    11-28 15:38:40.767 16137-16206/com.example.qidong.testrxjava E/MainActivity: 1:0
    11-28 15:38:40.767 16137-16206/com.example.qidong.testrxjava E/MainActivity: 1:1
    11-28 15:38:40.767 16137-16206/com.example.qidong.testrxjava E/MainActivity: 1:2
    

    打印的结果正如我们所想的那样。

    三、concatMap操作符
    concatMap操作符的作用和flatMap的作用差不多,不同点在于它的结果是严格按照被观察者发送事件的顺序。下面看一下代码:

     Observable.create(new ObservableOnSubscribe<Integer>() {
    
                @Override
                public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                    emitter.onNext(1);
                    emitter.onNext(2);
                    emitter.onNext(3);
    
                }
            }).concatMap(new Function<Integer, ObservableSource<String>>() {
                @Override
                public ObservableSource<String> apply(Integer integer) throws Exception {
                    List<String> list=new ArrayList<>();
                    for(int i=0;i<3;i++)
                    {
                        list.add(integer+":"+i);
                    }
                    //延迟5秒
                    return Observable.fromIterable(list).delay(10, TimeUnit.MILLISECONDS);
                }
            }).subscribe(new Consumer<String>() {
                @Override
                public void accept(String s) throws Exception {
                    Log.e(TAG,s);
                }
            });
    

    运行结果:

    11-28 15:57:45.873 18765-18835/com.example.qidong.testrxjava E/MainActivity: 1:0
    11-28 15:57:45.873 18765-18835/com.example.qidong.testrxjava E/MainActivity: 1:1
    11-28 15:57:45.874 18765-18835/com.example.qidong.testrxjava E/MainActivity: 1:2
    11-28 15:57:45.892 18765-18836/com.example.qidong.testrxjava E/MainActivity: 2:0
    11-28 15:57:45.893 18765-18836/com.example.qidong.testrxjava E/MainActivity: 2:1
    11-28 15:57:45.896 18765-18836/com.example.qidong.testrxjava E/MainActivity: 2:2
    11-28 15:57:45.908 18765-18839/com.example.qidong.testrxjava E/MainActivity: 3:0
    11-28 15:57:45.909 18765-18839/com.example.qidong.testrxjava E/MainActivity: 3:1
    11-28 15:57:45.910 18765-18839/com.example.qidong.testrxjava E/MainActivity: 3:2
    

    可以看到结果是有序的。

    四、实例应用
    以一个虚拟的注册登录请求为例,用RxJava结合Retrofit来实现这样的功能。

    public interface Api {
        @GET
        Observable<LoginResponse> login(@Body LoginRequest request);
    
        @GET
        Observable<RegisterResponse> register(@Body RegisterRequest request);
    }
    

    可以看到登录和注册返回的都是一个上游Observable, 而我们的flatMap操作符的作用就是把一个Observable转换为另一个Observable, 因此结果就很显而易见了:

                api.register(new RegisterRequest())            //发起注册请求
                    .subscribeOn(Schedulers.io())               //在IO线程进行网络请求
                    .observeOn(AndroidSchedulers.mainThread())  //回到主线程去处理请求注册结果
                    .doOnNext(new Consumer<RegisterResponse>() {
                        @Override
                        public void accept(RegisterResponse registerResponse) throws Exception {
                            //先根据注册的响应结果去做一些操作
                        }
                    })
                    .observeOn(Schedulers.io())                 //回到IO线程去发起登录请求
                    .flatMap(new Function<RegisterResponse, ObservableSource<LoginResponse>>() {
                        @Override
                        public ObservableSource<LoginResponse> apply(RegisterResponse registerResponse) throws Exception {
                            return api.login(new LoginRequest());
                        }
                    })
                    .observeOn(AndroidSchedulers.mainThread())  //回到主线程去处理请求登录的结果
                    .subscribe(new Consumer<LoginResponse>() {
                        @Override
                        public void accept(LoginResponse loginResponse) throws Exception {
                            Toast.makeText(MainActivity.this, "登录成功", Toast.LENGTH_SHORT).show();
                        }
                    }, new Consumer<Throwable>() {
                        @Override
                        public void accept(Throwable throwable) throws Exception {
                            Toast.makeText(MainActivity.this, "登录失败", Toast.LENGTH_SHORT).show();
                        }
                    });
    

    从这个例子也可以看到我们切换线程是多么简单.

    相关文章

      网友评论

          本文标题:RxJava的操作符

          本文链接:https://www.haomeiwen.com/subject/xtzaqqtx.html