如何使该rxjava zip并行运行?
问题内容:
我有一个模拟长时间运行的睡眠方法。
private void sleep() {
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
然后我有一个方法返回一个Observable,其中包含参数中给出的2个字符串的列表。它在返回字符串之前调用sleep。
private Observable<List<String>> getStrings(final String str1, final String str2) {
return Observable.fromCallable(new Callable<List<String>>() {
@Override
public List<String> call() {
sleep();
List<String> strings = new ArrayList<>();
strings.add(str1);
strings.add(str2);
return strings;
}
});
}
然后,我在Observalb.zip中三次调用getStrings,我希望这三个调用可以并行运行,所以执行的总时间应该在 2秒
以内,也许最多是3秒,因为睡眠只有2秒。但是,总共需要 六 秒钟。 我如何使它并行运行,以便在2秒内完成?
Observable
.zip(getStrings("One", "Two"), getStrings("Three", "Four"), getStrings("Five", "Six"), mergeStringLists())
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<List<String>>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(List<String> strings) {
//Display the strings
}
});
mergeStringLists方法
private Func3<List<String>, List<String>, List<String>, List<String>> mergeStringLists() {
return new Func3<List<String>, List<String>, List<String>, List<String>>() {
@Override
public List<String> call(List<String> strings, List<String> strings2, List<String> strings3) {
Log.d(TAG, "...");
for (String s : strings2) {
strings.add(s);
}
for (String s : strings3) {
strings.add(s);
}
return strings;
}
};
}
问题答案:
发生这种情况是因为订阅您的zipped
可观察性发生在同一io
线程中。
您为什么不试试这个呢?
Observable
.zip(
getStrings("One", "Two")
.subscribeOn(Schedulers.newThread()),
getStrings("Three", "Four")
.subscribeOn(Schedulers.newThread()),
getStrings("Five", "Six")
.subscribeOn(Schedulers.newThread()),
mergeStringLists())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<List<String>>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(List<String> strings) {
//Display the strings
}
});
让我知道是否有帮助