RxJava入门（3）：变换操作符

演示一下RxJava的变换操作符

map

被观察者发送的每个事件都被通过指定函数处理，从而变成另一种事件

 Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                    e.onNext(1);
                    e.onNext(2);
                    e.onNext(3);
            }
        }).map(new Function<Integer, String>() {
            @Override
            public String apply(Integer integer) throws Exception {
                Log.e("yzh","function");
                return "Map--apply--"+integer;
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e("yzh","onSubscribe");
            }

            @Override
            public void onNext(String s) {
                Log.e("yzh","onNext--"+s);
            }

            @Override
            public void onError(Throwable e) {
                Log.e("yzh","onError--"+e.toString());
            }

            @Override
            public void onComplete() {
                Log.e("yzh","onComlete");
            }
        });

打印结果,本身事件int参数变成了string

01-31 06:21:19.242 3894-3894/com.stone.appuser.rxtest E/yzh: onSubscribe
01-31 06:21:19.242 3894-3894/com.stone.appuser.rxtest E/yzh: function
01-31 06:21:19.242 3894-3894/com.stone.appuser.rxtest E/yzh: onNext--Map--apply--1
01-31 06:21:19.242 3894-3894/com.stone.appuser.rxtest E/yzh: function
01-31 06:21:19.242 3894-3894/com.stone.appuser.rxtest E/yzh: onNext--Map--apply--2
01-31 06:21:19.242 3894-3894/com.stone.appuser.rxtest E/yzh: function
01-31 06:21:19.242 3894-3894/com.stone.appuser.rxtest E/yzh: onNext--Map--apply--3

flatMap

将被观察者发送的事件序列进行拆分，转化，再合并成一个新的事件序列最后再进行发送

 Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                e.onNext(2);
                e.onNext(3);
            }
        }).flatMap(new Function<Integer, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(Integer integer) throws Exception {
                Log.e("yzh","apply");
                List<String> list =new ArrayList<>();
                for(int i=0;i<3;i++){
                    list.add("我是事件"+integer+"拆分的子事件"+i);
                }
                return Observable.fromIterable(list);
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e("yzh","onSubscribe");
            }

            @Override
            public void onNext(String s) {
                Log.e("yzh","onNext--"+s);
            }

            @Override
            public void onError(Throwable e) {
                Log.e("yzh","onError--"+e.toString());
            }

            @Override
            public void onComplete() {
                Log.e("yzh","onComplete");
            }
        });

打印结果
这里有的介绍里说在被观察中的事件会被重新无序的组合，但是实际上并没有发生,如下。可以在return Observable.fromIterable(list)后面加上delay(10,TimeUnit.MILLISECONDS)延时 会产生无序的效果,感兴趣的朋友可以自己试试。

01-31 06:54:21.558 5228-5228/com.stone.appuser.rxtest E/yzh: onSubscribe
01-31 06:54:21.558 5228-5228/com.stone.appuser.rxtest E/yzh: apply
01-31 06:54:21.560 5228-5228/com.stone.appuser.rxtest E/yzh: onNext--我是事件1拆分的子事件0
01-31 06:54:21.560 5228-5228/com.stone.appuser.rxtest E/yzh: onNext--我是事件1拆分的子事件1
01-31 06:54:21.560 5228-5228/com.stone.appuser.rxtest E/yzh: onNext--我是事件1拆分的子事件2
01-31 06:54:21.560 5228-5228/com.stone.appuser.rxtest E/yzh: apply
01-31 06:54:21.560 5228-5228/com.stone.appuser.rxtest E/yzh: onNext--我是事件2拆分的子事件0
01-31 06:54:21.561 5228-5228/com.stone.appuser.rxtest E/yzh: onNext--我是事件2拆分的子事件1
01-31 06:54:21.561 5228-5228/com.stone.appuser.rxtest E/yzh: onNext--我是事件2拆分的子事件2
01-31 06:54:21.561 5228-5228/com.stone.appuser.rxtest E/yzh: apply
01-31 06:54:21.561 5228-5228/com.stone.appuser.rxtest E/yzh: onNext--我是事件3拆分的子事件0
01-31 06:54:21.562 5228-5228/com.stone.appuser.rxtest E/yzh: onNext--我是事件3拆分的子事件1
01-31 06:54:21.562 5228-5228/com.stone.appuser.rxtest E/yzh: onNext--我是事件3拆分的子事件2

concatMap

作用与flatMap一样，区别在于concatMap会将被观察者中的事件拆分组合之后按照被观察者中的序列传送给观察者，不管加不加上延时

  Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                e.onNext(2);
                e.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);
                }
                return Observable.fromIterable(list);
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.e("yzh","accept--"+s);
            }
        });

打印结果

01-31 07:11:44.871 5829-5829/com.stone.appuser.rxtest E/yzh: accept--我是事件 1拆分后的子事件0
01-31 07:11:44.871 5829-5829/com.stone.appuser.rxtest E/yzh: accept--我是事件 1拆分后的子事件1
01-31 07:11:44.871 5829-5829/com.stone.appuser.rxtest E/yzh: accept--我是事件 1拆分后的子事件2
01-31 07:11:44.871 5829-5829/com.stone.appuser.rxtest E/yzh: accept--我是事件 2拆分后的子事件0
01-31 07:11:44.871 5829-5829/com.stone.appuser.rxtest E/yzh: accept--我是事件 2拆分后的子事件1
01-31 07:11:44.871 5829-5829/com.stone.appuser.rxtest E/yzh: accept--我是事件 2拆分后的子事件2
01-31 07:11:44.871 5829-5829/com.stone.appuser.rxtest E/yzh: accept--我是事件 3拆分后的子事件0
01-31 07:11:44.871 5829-5829/com.stone.appuser.rxtest E/yzh: accept--我是事件 3拆分后的子事件1
01-31 07:11:44.871 5829-5829/com.stone.appuser.rxtest E/yzh: accept--我是事件 3拆分后的子事件2

buffer

定期从 被观察者需要发送的事件中 获取一定数量的事件并且放到缓存区中，最终发送

Observable.just(1,2,3,4,5)
                //第一个参数设置缓存区大小，每次从被观察者中获取事件的数量
                //第二个参数设置每次获取新事件数量
                .buffer(3,1)
                .subscribe(new Observer<List<Integer>>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.e("yzh","onSubscribe");
                    }

                    @Override
                    public void onNext(List<Integer> integers) {
                        Log.e("yzh","缓存区里的事件数量=="+integers.size());
                        for(Integer value:integers){
                            Log.e("yzh","事件--"+value);
                        }
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });

打印结果

01-31 07:34:56.709 6593-6593/com.stone.appuser.rxtest E/yzh: onSubscribe
01-31 07:34:56.709 6593-6593/com.stone.appuser.rxtest E/yzh: 缓存区里的事件数量==3
01-31 07:34:56.709 6593-6593/com.stone.appuser.rxtest E/yzh: 事件--1
01-31 07:34:56.709 6593-6593/com.stone.appuser.rxtest E/yzh: 事件--2
01-31 07:34:56.709 6593-6593/com.stone.appuser.rxtest E/yzh: 事件--3
01-31 07:34:56.710 6593-6593/com.stone.appuser.rxtest E/yzh: 缓存区里的事件数量==3
01-31 07:34:56.710 6593-6593/com.stone.appuser.rxtest E/yzh: 事件--2
01-31 07:34:56.710 6593-6593/com.stone.appuser.rxtest E/yzh: 事件--3
01-31 07:34:56.710 6593-6593/com.stone.appuser.rxtest E/yzh: 事件--4
01-31 07:34:56.710 6593-6593/com.stone.appuser.rxtest E/yzh: 缓存区里的事件数量==3
01-31 07:34:56.710 6593-6593/com.stone.appuser.rxtest E/yzh: 事件--3
01-31 07:34:56.711 6593-6593/com.stone.appuser.rxtest E/yzh: 事件--4
01-31 07:34:56.711 6593-6593/com.stone.appuser.rxtest E/yzh: 事件--5
01-31 07:34:56.711 6593-6593/com.stone.appuser.rxtest E/yzh: 缓存区里的事件数量==2
01-31 07:34:56.711 6593-6593/com.stone.appuser.rxtest E/yzh: 事件--4
01-31 07:34:56.711 6593-6593/com.stone.appuser.rxtest E/yzh: 事件--5
01-31 07:34:56.711 6593-6593/com.stone.appuser.rxtest E/yzh: 缓存区里的事件数量==1
01-31 07:34:56.711 6593-6593/com.stone.appuser.rxtest E/yzh: 事件--5