此篇內(nèi)容均是來自書籍《RxJava響應(yīng)式編程》李衍順 著

3.8 聚合操作符

3.8.1 concat

concat操作符將多個(gè)Observable結(jié)合成一個(gè)Observable并發(fā)送數(shù)據(jù)，并且嚴(yán)格按照先后順序發(fā)送數(shù)據(jù)，即前一個(gè)Observable數(shù)據(jù)沒有發(fā)送完時(shí)，后面的Observable是不能發(fā)送數(shù)據(jù)的。

有兩個(gè)操作符與concat操作符很類似，他們分別是：
-startWith：僅僅是在前面插上一個(gè)Observable或者一些數(shù)據(jù)，并且先發(fā)送插入的內(nèi)容
-merge：其發(fā)送的數(shù)據(jù)是無序的，也就是說被組合的多個(gè)Observable是可以自由發(fā)送數(shù)據(jù)的，而不用管其它的Observable的狀態(tài)。

/**
     * concat
     */
    private void concatTest(){
        Observable<Integer> observable1 = Observable.just(1,2,3);
        Observable<Integer> observable2 = Observable.just(4,5,6);
        Observable<Integer> observable3 = Observable.just(7,8,9);
        Observable.concat(observable1, observable3, observable2)
                .subscribe(new Action1<Integer>() {
                    @Override
                    public void call(Integer integer) {
                        log("concat: " + integer);
                    }
                });
    }
結(jié)果：
concat: 1
concat: 2
concat: 3
concat: 7
concat: 8
concat: 9
concat: 4
concat: 5
concat: 6

3.8.2 count

count操作符用來統(tǒng)計(jì)源Observable發(fā)送了多少個(gè)數(shù)據(jù)，最后將數(shù)目發(fā)送出來。如果源Observable發(fā)送錯(cuò)誤，則會(huì)將錯(cuò)誤直接報(bào)出來。在源Observable停止發(fā)送前，count是不會(huì)發(fā)送統(tǒng)計(jì)數(shù)據(jù)的

/**
     * count
     */
    private void countTest(){
        Observable.just(1,2,3).count()
                .subscribe(new Action1<Integer>() {
                    @Override
                    public void call(Integer integer) {
                        log("count: " + integer);
                    }
                });
    }
結(jié)果：
count: 3

3.8.3 reduce

reduce操作符應(yīng)用一個(gè)函數(shù)接收Observable發(fā)送的數(shù)據(jù)和函數(shù)的計(jì)算結(jié)果，作為下次計(jì)算的參數(shù)，并輸出最后的結(jié)果。reduce與我們前面了解過的scan操作符很類似，只是scan會(huì)輸出每次計(jì)算的結(jié)果，而reduce只輸出最后的結(jié)果。

/**
     * reduce
     */
    private void reduceTest(){
        ArrayList<Integer> list = new ArrayList<>();
        for(int i=0; i<10;i++){
            list.add(2);
        }
        Observable.from(list).reduce(new Func2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer integer, Integer integer2) {
                return integer * integer2;
            }
        }).subscribe(new Action1<Integer>() {
            @Override
            public void call(Integer integer) {
                log("reduce: " + integer);
            }
        });
    }
結(jié)果：
reduce: 1024

3.8.4 collect

collect操作符類似于reduce，但是二者目的不同。collect操作符用來將源Observable發(fā)送的數(shù)據(jù)收集到一個(gè)數(shù)據(jù)結(jié)構(gòu)里面，最后將這個(gè)數(shù)據(jù)結(jié)構(gòu)整個(gè)發(fā)送出來。collect操作符需要使用兩個(gè)函數(shù)作為參數(shù)：
-第一個(gè)函數(shù)會(huì)產(chǎn)生收集數(shù)據(jù)結(jié)構(gòu)的函數(shù)
-第二個(gè)函數(shù)會(huì)將上面函數(shù)產(chǎn)生的數(shù)據(jù)結(jié)構(gòu)和源Observable發(fā)送的數(shù)據(jù)作為參數(shù)，且這個(gè)函數(shù)會(huì)將源Observable發(fā)送的數(shù)據(jù)存入到這個(gè)數(shù)據(jù)結(jié)構(gòu)中。

/**
     * collect
     */
    private void collectTest(){
        ArrayList<Integer> list = new ArrayList<>();
        for(int i=0; i<10;i++){
            list.add(i);
        }
        Observable.from(list).collect(new Func0<ArrayList<Integer>>() {
            @Override
            public ArrayList<Integer> call() {
                return new ArrayList<>();
            }
        }, new Action2<ArrayList<Integer>, Integer>() {
            @Override
            public void call(ArrayList<Integer> integers, Integer integer) {
                integers.add(integer);
            }
        }).subscribe(new Action1<ArrayList<Integer>>() {
            @Override
            public void call(ArrayList<Integer> integers) {
                log("collect: " + integers);
            }
        });
    }
結(jié)果：
collect: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]