Rxjava-组合类操作符 · dangxy99的Blog

组合类操作符(merge zip join combineLatest and/when/then switch startSwitch…)

range()

你需要从一个指定的数字X开始发射N个数字吗？你可以用range:

Observable.range(10,3)
    .subscribe(new Observable<Integer>() {
        @Override
        public void onCompleted() {
            Toast.makeText(getActivity(), "Yeaaah!", Toast.LENGTH_LONG).show();
        }
        @Override
        public void onError(Throwable e) {
            Toast.makeText(getActivity(), "Something went wrong!", Toast.LENGTH_SHORT).show();
        }
        @Override
        public void onNext(Integer number) {
            Toast.makeText(getActivity(), "I say " + number, Toast.LENGTH_SHORT).show();
        }
    });
range()函数用两个数字作为参数：第一个是起始点，第二个是我们想发射数字的个数。

interval

interval()函数在你需要创建一个轮询程序时非常好用。
interval()函数的两个参数：一个指定两次发射的时间间隔，另一个是用到的时间单位。

Subscription stopMePlease = Observable.interval(3,TimeUnit.SECONDS)
    .subscribe(new Observable<Integer>() {
        @Override
        public void onCompleted() {
            Toast.makeText(getActivity(), "Yeaaah!", Toast.LENGTH_LONG).show();
        }
        @Override
        public void onError(Throwable e) {
            Toast.makeText(getActivity(), "Something went wrong!", Toast.LENGTH_SHORT).show();
        }
        @Override
        public void onNext(Integer number) {
            Toast.makeText(getActivity(), "I say " + number, Toast.LENGTH_SHORT).show();
        }
    });

timer()

如果你需要一个一段时间之后才发射的Observable，你可以像下面的例子使用timer()：

Observable.timer(3,TimeUnit.SECONDS)
    .subscribe(new Observable<Long>() {
        @Override
        public void onCompleted() {
            
        }
        @Override
        public void onError(Throwable e) {
            
        }
        @Override
        public void onNext(Long number) {
            Log.d("RXJAVA", "I say " + number);
        }
    });
它将3秒后发射0,然后就完成了。让我们使用timer()的第三个参数，就像下面的例子：
Observable.timer(3,3,TimeUnit.SECONDS)
    .subscribe(new Observable<Long>() {
        @Override
        public void onCompleted() {
            
        }
        @Override
        public void onError(Throwable e) {
            
        }
        @Override
        public void onNext(Long number) {
            Log.d("RXJAVA", "I say " + number);
        }
    });

Merge

merge(Observable, Observable)将两个Observable发射的事件序列组合并成一个事件序列，就像是一个Observable发射的一样。你可以简单的将它理解为两个Obsrvable合并成了一个Observable，合并后的数据是无序的。

String[] letters = new String[]{"A", "B", "C", "D", "E", "F", "G", "H"};
Observable<String> letterSequence = Observable.interval(300, TimeUnit.MILLISECONDS)
        .map(new Func1<Long, String>() {
            @Override
            public String call(Long position) {
                return letters[position.intValue()];
            }
        }).take(letters.length);
Observable<Long> numberSequence = Observable.interval(500, TimeUnit.MILLISECONDS).take(5);
Observable.merge(letterSequence, numberSequence)
        .subscribe(new Observer<Serializable>() {
            @Override
            public void onCompleted() {
                System.exit(0);
            }
            @Override
            public void onError(Throwable e) {
                System.out.println("Error:" + e.getMessage());
            }
            @Override
            public void onNext(Serializable serializable) {
                System.out.print(serializable.toString()+" ");
            }
        });

merge(Observable[])将多个Observable发射的事件序列组合并成一个事件序列，就像是一个Observable发射的一样。

StartWith

startWith(T)用于在源Observable发射的数据前插入数据。使用startWith(Iterable)我们还可以在源Observable发射的数据前插入Iterable。官方示意图：

startWith(Observable)用于在源Observable发射的数据前插入另一个Observable发射的数据（这些数据会被插入到源Observable发射数据的前面）。官方示意图：

Concat

concat(Observable<? extends T>, Observable<? extends T>) concat(Observable<？ extends Observable>)用于将多个obserbavle发射的的数据进行合并发射，concat严格按照顺序发射数据，前一个Observable没发射玩是不会发射后一个Observable的数据的。它和merge、startWitch和相似，不同之处在于：

merge:合并后发射的数据是无序的；

startWitch:只能在源Observable发射的数据前插入数据。

String[] letters = new String[]{"A", "B", "C", "D", "E", "F", "G", "H"};
Observable<String> letterSequence = Observable.interval(300, TimeUnit.MILLISECONDS)
        .map(new Func1<Long, String>() {
            @Override
            public String call(Long position) {
                return letters[position.intValue()];
            }
        }).take(letters.length);
Observable<Long> numberSequence = Observable.interval(500, TimeUnit.MILLISECONDS).take(5);
Observable.concat(letterSequence, numberSequence)
        .subscribe(new Observer<Serializable>() {
            @Override
            public void onCompleted() {
                System.exit(0);
            }
            @Override
            public void onError(Throwable e) {
                System.out.println("Error:" + e.getMessage());
            }
            @Override
            public void onNext(Serializable serializable) {
                System.out.print(serializable.toString() + " ");
            }
        });

ZIP

zip(Observable, Observable, Func2)用来合并两个Observable发射的数据项，根据Func2函数生成一个新的值并发射出去。当其中一个Observable发送数据结束或者出现异常后，另一个Observable也将停在发射数据。

String[] letters = new String[]{"A", "B", "C", "D", "E", "F", "G", "H"};
Observable<String> letterSequence = Observable.interval(120, TimeUnit.MILLISECONDS)
        .map(new Func1<Long, String>() {
            @Override
            public String call(Long position) {
                return letters[position.intValue()];
            }
        }).take(letters.length);
Observable<Long> numberSequence = Observable.interval(200, TimeUnit.MILLISECONDS).take(5);
Observable.zip(letterSequence, numberSequence, new Func2<String, Long, String>() {
    @Override
    public String call(String letter, Long number) {
        return letter + number;
    }
}).subscribe(new Observer<String>() {
    @Override
    public void onCompleted() {
        System.exit(0);
    }
    @Override
    public void onError(Throwable e) {
        System.out.println("Error:" + e.getMessage());
    }
    @Override
    public void onNext(String result) {
        System.out.print(result + " ");
    }
});

CombineLatest

combineLatest(Observable, Observable, Func2)用于将两个Observale最近发射的数据已经Func2函数的规则进展组合。下面是官方提供的原理图：

List<String> communityNames = DataSimulator.getCommunityNames();
List<Location> locations = DataSimulator.getLocations();
Observable<String> communityNameSequence = Observable.interval(1, TimeUnit.SECONDS)
        .map(new Func1<Long, String>() {
            @Override
            public String call(Long position) {
                return communityNames.get(position.intValue());
            }
        }).take(communityNames.size());
Observable<Location> locationSequence = Observable.interval(1, TimeUnit.SECONDS)
        .map(new Func1<Long, Location>() {
            @Override
            public Location call(Long position) {
                return locations.get(position.intValue());
            }
        }).take(locations.size());
Observable.combineLatest(
        communityNameSequence,
        locationSequence,
        new Func2<String, Location, String>() {
            @Override
            public String call(String communityName, Location location) {
                return "小区名:" + communityName + ", 经纬度:" + location.toString();
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onCompleted() {
                System.exit(0);
            }
            @Override
            public void onError(Throwable e) {
                System.out.println("Error:" + e.getMessage());
            }
            @Override
            public void onNext(String s) {
                System.out.println(s);
            }
        });
//输出
小区名:竹园新村, 经纬度:(21.827, 23.323)
小区名:康桥半岛, 经纬度:(21.827, 23.323)
小区名:康桥半岛, 经纬度:(11.923, 16.309)
小区名:中粮·海景壹号, 经纬度:(11.923, 16.309)
小区名:中粮·海景壹号, 经纬度:(22.273, 53.623)
小区名:浦江名苑, 经纬度:(22.273, 53.623)
小区名:南辉小区, 经纬度:(22.273, 53.623)

SwitchOnNext

switchOnNext(Observable<? extends Observable<? extends T>>用来将一个发射多个小Observable的源Observable转化为一个Observable，然后发射这多个小Observable所发射的数据。如果一个小的Observable正在发射数据的时候，源Observable又发射出一个新的小Observable，则前一个Observable发射的数据会被抛弃，直接发射新的小Observable所发射的数据。

结合下面的原理图大家应该很容易理解，我们可以看到下图中的黄色圆圈就被丢弃了。

Join

join(Observable, Func1, Func1, Func2)我们先介绍下join操作符的4个参数：

Observable：源Observable需要组合的Observable,这里我们姑且称之为目标Observable；
Func1：接收从源Observable发射来的数据，并返回一个Observable，这个Observable的声明周期决定了源Obsrvable发射出来的数据的有效期；
Func1：接收目标Observable发射来的数据，并返回一个Observable，这个Observable的声明周期决定了目标Obsrvable发射出来的数据的有效期；
Func2：接收从源Observable和目标Observable发射出来的数据，并将这两个数据组合后返回
所以Join操作符的语法结构大致是这样的：onservableA.join(observableB, 控制observableA发射数据有效期的函数，控制observableB发射数据有效期的函数，两个observable发射数据的合并规则)

join操作符的效果类似于排列组合，把第一个数据源A作为基座窗口，他根据自己的节奏不断发射数据元素，第二个数据源B，每发射一个数据，我们都把它和第一个数据源A中已经发射的数据进行一对一匹配；举例来说，如果某一时刻B发射了一个数据“B”,此时A已经发射了0，1，2，3共四个数据，那么我们的合并操作就会把“B”依次与0,1,2,3配对，得到四组数据： [0, B][1, B] [2, B] [3, B]

final List<House> houses = DataSimulator.getHouses();//模拟的房源数据，用于测试
//用来每秒从houses总取出一套房源并发射出去
Observable<House> houseSequence =
        Observable.interval(1, TimeUnit.SECONDS)
                .map(new Func1<Long, House>() {
                    @Override
                    public House call(Long position) {
                        return houses.get(position.intValue());
                    }
                }).take(houses.size());//这里的take是为了防止houses.get(position.intValue())数组越界
//用来实现每秒发送一个新的Long型数据
Observable<Long> tictoc = Observable.interval(1, TimeUnit.SECONDS);
houseSequence.join(tictoc,
        new Func1<House, Observable<Long>>() {
            @Override
            public Observable<Long> call(House house) {
                return Observable.timer(2, TimeUnit.SECONDS);
            }
        },
        new Func1<Long, Observable<Long>>() {
            @Override
            public Observable<Long> call(Long aLong) {
                return Observable.timer(0, TimeUnit.SECONDS);
            }
        },
        new Func2<House, Long, String>() {
            @Override
            public String call(House house, Long aLong) {
                return aLong + "-->" + house.getDesc();
            }
        }
).subscribe(new Observer<String>() {
    @Override
    public void onCompleted() {
        System.exit(0);
    }
    @Override
    public void onError(Throwable e) {
        System.out.println("Error:"+e.getMessage());
    }
    @Override
    public void onNext(String s) {
        System.out.println(s);
    }
});
//输入
0-->中粮海景壹号新出大平层！总价4500W起
1-->中粮海景壹号新出大平层！总价4500W起
1-->满五唯一，黄金地段
2-->中粮海景壹号新出大平层！总价4500W起
2-->满五唯一，黄金地段
2-->一楼自带小花园
3-->一楼自带小花园
3-->毗邻汤臣一品
4-->毗邻汤臣一品
4-->顶级住宅，给您总统般尊贵体验
5-->顶级住宅，给您总统般尊贵体验
5-->顶层户型，两室一厅
6-->顶层户型，两室一厅
6-->南北通透，豪华五房
7-->南北通透，豪华五房