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

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

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

前言

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

操作符总览

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

具体使用介绍

Buffer

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

代码示例1：批量收集

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
//一次订阅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());
    }
});

运行结果：

批量收集

代码示例：周期收集

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
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）~

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
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);
    }
});

运行结果：

flatMap

Map

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

代码示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
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);
    }
});

运行结果：

map

GroupBy

代码示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
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);
            }
        });
    }
});

运行结果：

GroupBy

Sacn

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

示例代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
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);
    }
});

运行结果：

sacn

Window

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

示例代码：

1
2
3
4
5
6
7
8
9
10
11
12
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);
            }
        });
    }
});

运行结果：

window

结语

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