RxJava进阶二（转换类操作符）

Posted 2022-05-31 王永迪

RxJava进阶一（创建类操作符）
RxJava进阶二（转换类操作符）
RxJava进阶三（过滤类操作符）
RxJava进阶四（组合类操作符）

---

前言

前面我们了解了RXJava的创建类操作符，这篇文章带着大家熟悉一下RxJava的转换类操作符，本系列文章仅是带大家认识一下这些操作符的用法，并没有对操作符进行多种形态的使用，具体的还要大家在使用时注意~

操作符总览

Buffer、FlatMap、GroupBy、GroupBy、Map、Scan、Window…

具体使用介绍

Buffer

可以简单的理解为缓存，它可以批量或者按周期性从Observable收集数据到一个集合，然后把这些数据集合打包发射，而不是一次发射一个数据~

代码示例1：批量收集

        //一次订阅2个
        Observable.range(1, 5).buffer(2).subscribe(new Subscriber<List<Integer>>() 
            @Override
            public void onCompleted() 
                System.out.println("onCompleted");
            

            @Override
            public void onError(Throwable e) 
                System.out.println("onError");
            

            @Override
            public void onNext(List<Integer> integers) 
                System.out.println("onNext" + integers.toString());
            
        );

运行结果：

代码示例：周期收集

        Observable.create(new Observable.OnSubscribe<String>() 
            @Override
            public void call(Subscriber<? super String> subscriber) 
                while (true) 
                    subscriber.onNext("消息" + System.currentTimeMillis());
                    SystemClock.sleep(2000);//每隔2s发送消息
                

            
        ).subscribeOn(Schedulers.io()).buffer(3, TimeUnit.SECONDS).subscribe(new Observer<List<String>>() 
            @Override
            public void onCompleted() 
                System.out.println("onCompleted");
            

            @Override
            public void onError(Throwable e) 
                System.out.println("onError");
            

            @Override
            public void onNext(List<String> integers) 
                System.out.println("onNext" + integers.toString());
            
        );

FlatMap

扁平映射，作用是将一个原始Observable发射的数据进行变化，输出一个或多个Observable，然后将这些Observable发射的数据平坦化的放进一个单独的Observable（参数一般是Func1）~

        List<String> stringList = new ArrayList<>();
        stringList.add("one");
        stringList.add("two");
        stringList.add("three");
        Observable.just(stringList).flatMap(new Func1<List<String>, Observable<String>>() 
            @Override
            public Observable<String> call(List<String> list) 
                return Observable.from(list);
            
        ).subscribe(new Action1<String>() 
            @Override
            public void call(String next) 
                System.out.println("onNext -->" + next);
            
        );

运行结果：

Map

映射，一般用于对原始的数据进行加工处理，返回一个加工过后的数据~

代码示例：

        List<String> stringList = new ArrayList<>();
        stringList.add("one");
        stringList.add("two");
        stringList.add("three");
        Observable.just(stringList).map(new Func1<List<String>, String>() 
            @Override
            public String call(List<String> strings) 
                return strings.toString();
            
        ).subscribe(new Action1<String>() 
            @Override
            public void call(String next) 
                System.out.println("onNext -->" + next);
            
        );

运行结果：

GroupBy

代码示例：

        Observable.range(0, 4).groupBy(new Func1<Integer, Integer>() 
            @Override
            public Integer call(Integer integer) 
                return integer % 2;//分成基数偶数两组
            
        ).subscribe(new Action1<GroupedObservable<Integer, Integer>>() 
            @Override
            public void call(GroupedObservable<Integer, Integer> integerIntegerGroupedObservable) 
                integerIntegerGroupedObservable.subscribe(new Action1<Integer>() 
                    @Override
                    public void call(Integer integer) 
                        System.out.println("Group -->" + integerIntegerGroupedObservable.getKey() + "  &&  OnNext -->" + integer);
                    
                );
            
        );

运行结果：

Sacn

sacn操作符是遍历源Observable产生的结果，通过自定义转换规则，依次输出结果给订阅者，

示例代码：

        Observable.range(0, 4).scan(new Func2<Integer, Integer, Integer>() 
            // 第一个参数是上次的结算结果，
            // 第二个参数是当此的源observable的输入值
            @Override
            public Integer call(Integer lastResult, Integer currentInput) 
                return lastResult + currentInput;
            
        ).subscribe(new Action1<Integer>() 
            @Override
            public void call(Integer result) 
                System.out.println("onNext -->" + result);
            
        );

运行结果：

Window

窗口，它可以批量或者按周期性从Observable收集数据到一个集合，然后把这些数据集合打包发射，而不是一次发射一个数据,类似于Buffer，但Buffer发射的是数据，Window发射的是Observable~

示例代码：

        Observable.range(1, 5).window(2).subscribe(new Action1<Observable<Integer>>() 
            @Override
            public void call(Observable<Integer> integerObservable) 
                System.out.println("onOutsideNext -->" + integerObservable);
                integerObservable.subscribe(new Action1<Integer>() 
                    @Override
                    public void call(Integer integer) 
                        System.out.println("onInsideNext -->" + integer);
                    
                );
            
        );

运行结果：

结语

转换类操作符，就简单介绍到这里，希望能够对同学有所帮助，谢谢~