我一直在寻找一个简单的答案,关于如何使用Zip操作符以及如何处理创建的Observables以将它们传递给Zip操作符。我想知道是否应该为每个Observable调用subscribe()方法。但是没有一个简单的答案可以找到,我不得不自己摸索。下面是一个使用Zip操作符处理两个Observables的简单示例:
@Test
public void zipOperator() throws Exception {
List<Integer> indexes = Arrays.asList(0, 1, 2, 3, 4);
List<String> letters = Arrays.asList("a", "b", "c", "d", "e");
Observable<Integer> indexesObservable = Observable.fromIterable(indexes);
Observable<String> lettersObservable = Observable.fromIterable(letters);
Observable.zip(indexesObservable, lettersObservable, mergeEmittedItems())
.subscribe(printMergedItems());
}
@NonNull
private BiFunction<Integer, String, String> mergeEmittedItems() {
return new BiFunction<Integer, String, String>() {
@Override
public String apply(Integer index, String letter) throws Exception {
return "[" + index + "] " + letter;
}
};
}
@NonNull
private Consumer<String> printMergedItems() {
return new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
System.out.println(s);
}
};
}
打印结果为:
[0] a
[1] b
[2] c
[3] d
[4] e
我脑海中的问题的最终答案如下:
传递给zip()方法的Observables只需要创建即可,它们不需要任何订阅者,仅仅创建就足够了......如果你想让任何Observable在调度程序上运行,可以为该Observable指定这个功能......我还尝试在Observables上使用zip()操作符,它们应该等待结果,当两个结果都准备好时,zip()的消费者才会被触发(这是预期的行为)。