Rxjava Android Zip操作符的使用方法
Rxjava Android how to use the Zip operator
我在理解 android 项目的 RxJava 中的 zip 运算符时遇到了很多麻烦。
问题
我需要能够发送网络请求来上传视频
然后我需要发送一个网络请求来上传一张图片来配合它
最后我需要添加一个描述并使用前两个请求的响应将视频和图片的位置 url 连同描述一起上传到我的服务器。
我认为 zip 运算符非常适合这项任务，因为我知道我们可以获取两个可观察对象（视频和图片请求）的响应并将它们用于我的最终任务。
但我似乎无法按照我的设想实现它。
我正在找人回答如何使用一些伪代码在概念上完成此操作。
谢谢
zip 运算符允许您从两个不同的可观察值的结果中组合出一个结果。
你必须给出一个 lambda，它将根据每个可观察对象发出的数据创建一个结果。
Observable<MovieResponse> movies = ...
Observable<PictureResponse> picture = ...
Observable<Response> response = movies.zipWith(picture, (movie, pic) -> {
return new Response("description", movie.getName(), pic.getUrl());
});
Zip 运算符严格配对来自 observable 的发射项。它等待两个（或更多）项目到达然后合并它们。所以是的，这将适合您的需求。
我会使用 Func2 链接前两个可观察量的结果。
请注意，如果您使用 Retrofit，此方法会更简单，因为它的 api 接口可能 return 是一个可观察对象。否则，您将需要创建自己的可观察对象。
// assuming each observable returns response in the form of String
Observable<String> movOb = Observable.create(...);
// if you use Retrofit
Observable<String> picOb = RetrofitApiManager.getService().uploadPic(...),
Observable.zip(movOb, picOb, new Func2<String, String, MyResult>() {
@Override
public MyResult call(String movieUploadResponse, String picUploadResponse) {
// analyze both responses, upload them to another server
// and return this method with a MyResult type
return myResult;
}
}
)
// continue chaining this observable with subscriber
// or use it for something else
这里我有一个以异步方式使用 Zip 的例子，以防你好奇
/**
* Since every observable into the zip is created to subscribeOn a diferent thread, it´s means all of them will run in parallel.
* By default Rx is not async, only if you explicitly use subscribeOn.
*/
@Test
public void testAsyncZip() {
scheduler = Schedulers.newThread();
scheduler1 = Schedulers.newThread();
scheduler2 = Schedulers.newThread();
long start = System.currentTimeMillis();
Observable.zip(obAsyncString(), obAsyncString1(), obAsyncString2(), (s, s2, s3) -> s.concat(s2)
.concat(s3))
.subscribe(result -> showResult("Async in:", start, result));
}
/**
* In this example the the three observables will be emitted sequentially and the three items will be passed to the pipeline
*/
@Test
public void testZip() {
long start = System.currentTimeMillis();
Observable.zip(obString(), obString1(), obString2(), (s, s2, s3) -> s.concat(s2)
.concat(s3))
.subscribe(result -> showResult("Sync in:", start, result));
}
public void showResult(String transactionType, long start, String result) {
System.out.println(result + " " +
transactionType + String.valueOf(System.currentTimeMillis() - start));
}
public Observable<String> obString() {
return Observable.just("")
.doOnNext(val -> {
System.out.println("Thread " + Thread.currentThread()
.getName());
})
.map(val -> "Hello");
}
public Observable<String> obString1() {
return Observable.just("")
.doOnNext(val -> {
System.out.println("Thread " + Thread.currentThread()
.getName());
})
.map(val -> " World");
}
public Observable<String> obString2() {
return Observable.just("")
.doOnNext(val -> {
System.out.println("Thread " + Thread.currentThread()
.getName());
})
.map(val -> "!");
}
public Observable<String> obAsyncString() {
return Observable.just("")
.observeOn(scheduler)
.doOnNext(val -> {
System.out.println("Thread " + Thread.currentThread()
.getName());
})
.map(val -> "Hello");
}
public Observable<String> obAsyncString1() {
return Observable.just("")
.observeOn(scheduler1)
.doOnNext(val -> {
System.out.println("Thread " + Thread.currentThread()
.getName());
})
.map(val -> " World");
}
public Observable<String> obAsyncString2() {
return Observable.just("")
.observeOn(scheduler2)
.doOnNext(val -> {
System.out.println("Thread " + Thread.currentThread()
.getName());
})
.map(val -> "!");
}
您可以在此处查看更多示例https://github.com/politrons/reactive
一个小例子：
val observableOne = Observable.just("Hello", "World")
val observableTwo = Observable.just("Bye", "Friends")
val zipper = BiFunction<String, String, String> { first, second -> "$first - $second" }
Observable.zip(observableOne, observableTwo, zipper)
.subscribe { println(it) }
这将打印：
Hello - Bye
World - Friends
在BiFunction<String, String, String>中第一个String第一个observable的类型，第二个String是第二个observable的类型，第三个String表示类型你的拉链函数的return。
我做了一个小例子，在 this blog post
中使用 zip 调用了两个真实端点
我一直在寻找关于如何使用 Zip 运算符的简单答案，以及如何处理我创建的 Observables 以将它们传递给它，我想知道我是否应该为每个 observable 或不，这些答案都不是很容易找到的，我必须自己弄清楚，所以这是一个在 2 个 Observables 上使用 Zip 运算符的简单示例：
@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 指定这个 ...这是预期的行为）
这是我使用 Single.zip 和 rxJava2
的实现
我尽量让它简单易懂
//
// API Client Interface
//
@GET(ServicesConstants.API_PREFIX + "questions/{id}/")
Single<Response<ResponseGeneric<List<ResponseQuestion>>>> getBaseQuestions(@Path("id") int personId);
@GET(ServicesConstants.API_PREFIX + "physician/{id}/")
Single<Response<ResponseGeneric<List<ResponsePhysician>>>> getPhysicianInfo(@Path("id") int personId);
//
// API middle layer - NOTE: I had feedback that the Single.create is not needed (but I haven't yet spent the time to improve it)
//
public Single<List<ResponsePhysician>> getPhysicianInfo(int personId) {
return Single.create(subscriber -> {
apiClient.getPhysicianInfo(appId)
.subscribeOn(Schedulers.io())
.observeOn(Schedulers.io())
.subscribe(response -> {
ResponseGeneric<List<ResponsePhysician>> responseBody = response.body();
if(responseBody != null && responseBody.statusCode == 1) {
if (!subscriber.isDisposed()) subscriber.onSuccess(responseBody.data);
} else if(response.body() != null && response.body().status != null ){
if (!subscriber.isDisposed()) subscriber.onError(new Throwable(response.body().status));
} else {
if (!subscriber.isDisposed()) subscriber.onError(new Throwable(response.message()));
}
}, throwable -> {
throwable.printStackTrace();
if(!subscriber.isDisposed()) subscriber.onError(throwable);
});
});
}
public Single<List<ResponseQuestion>> getHealthQuestions(int personId){
return Single.create(subscriber -> {
apiClient.getBaseQuestions(personId)
.subscribeOn(Schedulers.io())
.observeOn(Schedulers.io())
.subscribe(response -> {
ResponseGeneric<List<ResponseQuestion>> responseBody = response.body();
if(responseBody != null && responseBody.data != null) {
if (!subscriber.isDisposed()) subscriber.onSuccess(response.body().data);
} else if(response.body() != null && response.body().status != null ){
if (!subscriber.isDisposed()) subscriber.onError(new Throwable(response.body().status));
} else {
if (!subscriber.isDisposed()) subscriber.onError(new Throwable(response.message()));
}
}, throwable -> {
throwable.printStackTrace();
if(!subscriber.isDisposed()) subscriber.onError(throwable);
});
});
}
//please note that ResponseGeneric is just an outer wrapper of the returned data - common to all API's in this project
public class ResponseGeneric<T> {
@SerializedName("Status")
public String status;
@SerializedName("StatusCode")
public float statusCode;
@SerializedName("Data")
public T data;
}
//
// API end-use layer - this gets close to the UI so notice the oberver is set for main thread
//
private static class MergedResponse{// this is just a POJO to store all the responses in one object
public List<ResponseQuestion> listQuestions;
public List<ResponsePhysician> listPhysicians;
public MergedResponse(List<ResponseQuestion> listQuestions, List<ResponsePhysician> listPhysicians){
this.listQuestions = listQuestions;
this.listPhysicians = listPhysicians;
}
}
// example of Single.zip() - calls getHealthQuestions() and getPhysicianInfo() from API Middle Layer
private void downloadHealthQuestions(int personId) {
addRxSubscription(Single
.zip(getHealthQuestions(personId), getPhysicianInfo(personId), MergedResponse::new)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(response -> {
if(response != null) {
Timber.i(" - total health questions downloaded %d", response.listQuestions.size());
Timber.i(" - physicians downloaded %d", response.listPhysicians.size());
if (response.listPhysicians != null && response.listPhysicians.size()>0) {
// do your stuff to process response data
}
if (response.listQuestions != null && response.listQuestions.size()>0) {
// do your stuff to process response data
}
} else {
// process error - show message
}
}, error -> {
// process error - show network error message
}));
}
您将 rxjava 中的 zip 与 Java 8 一起使用：
Observable<MovieResponse> movies = ...
Observable<PictureResponse> picture = ...
Observable<ZipResponse> response = Observable.zip(movies, picture, ZipResponse::new);
class ZipResponse {
private MovieResponse movieResponse;
private PictureResponse pictureResponse;
ZipResponse(MovieResponse movieResponse, PictureResponse pictureResponse) {
this.movieResponse = movieResponse;
this.pictureResponse = pictureResponse;
}
public MovieResponse getMovieResponse() {
return movieResponse;
}
public void setMovieResponse(MovieResponse movieResponse) {
this.movieResponse= movieResponse;
}
public PictureResponse getPictureResponse() {
return pictureResponse;
}
public void setPictureResponse(PictureResponse pictureResponse) {
this.pictureResponse= pictureResponse;
}
}
我在理解 android 项目的 RxJava 中的 zip 运算符时遇到了很多麻烦。 问题 我需要能够发送网络请求来上传视频 然后我需要发送一个网络请求来上传一张图片来配合它 最后我需要添加一个描述并使用前两个请求的响应将视频和图片的位置 url 连同描述一起上传到我的服务器。
我认为 zip 运算符非常适合这项任务，因为我知道我们可以获取两个可观察对象（视频和图片请求）的响应并将它们用于我的最终任务。 但我似乎无法按照我的设想实现它。
我正在找人回答如何使用一些伪代码在概念上完成此操作。 谢谢
zip 运算符允许您从两个不同的可观察值的结果中组合出一个结果。
你必须给出一个 lambda，它将根据每个可观察对象发出的数据创建一个结果。
Observable<MovieResponse> movies = ...
Observable<PictureResponse> picture = ...
Observable<Response> response = movies.zipWith(picture, (movie, pic) -> {
return new Response("description", movie.getName(), pic.getUrl());
});
Zip 运算符严格配对来自 observable 的发射项。它等待两个（或更多）项目到达然后合并它们。所以是的，这将适合您的需求。
我会使用 Func2 链接前两个可观察量的结果。
请注意，如果您使用 Retrofit，此方法会更简单，因为它的 api 接口可能 return 是一个可观察对象。否则，您将需要创建自己的可观察对象。
// assuming each observable returns response in the form of String
Observable<String> movOb = Observable.create(...);
// if you use Retrofit
Observable<String> picOb = RetrofitApiManager.getService().uploadPic(...),
Observable.zip(movOb, picOb, new Func2<String, String, MyResult>() {
@Override
public MyResult call(String movieUploadResponse, String picUploadResponse) {
// analyze both responses, upload them to another server
// and return this method with a MyResult type
return myResult;
}
}
)
// continue chaining this observable with subscriber
// or use it for something else
这里我有一个以异步方式使用 Zip 的例子，以防你好奇
/**
* Since every observable into the zip is created to subscribeOn a diferent thread, it´s means all of them will run in parallel.
* By default Rx is not async, only if you explicitly use subscribeOn.
*/
@Test
public void testAsyncZip() {
scheduler = Schedulers.newThread();
scheduler1 = Schedulers.newThread();
scheduler2 = Schedulers.newThread();
long start = System.currentTimeMillis();
Observable.zip(obAsyncString(), obAsyncString1(), obAsyncString2(), (s, s2, s3) -> s.concat(s2)
.concat(s3))
.subscribe(result -> showResult("Async in:", start, result));
}
/**
* In this example the the three observables will be emitted sequentially and the three items will be passed to the pipeline
*/
@Test
public void testZip() {
long start = System.currentTimeMillis();
Observable.zip(obString(), obString1(), obString2(), (s, s2, s3) -> s.concat(s2)
.concat(s3))
.subscribe(result -> showResult("Sync in:", start, result));
}
public void showResult(String transactionType, long start, String result) {
System.out.println(result + " " +
transactionType + String.valueOf(System.currentTimeMillis() - start));
}
public Observable<String> obString() {
return Observable.just("")
.doOnNext(val -> {
System.out.println("Thread " + Thread.currentThread()
.getName());
})
.map(val -> "Hello");
}
public Observable<String> obString1() {
return Observable.just("")
.doOnNext(val -> {
System.out.println("Thread " + Thread.currentThread()
.getName());
})
.map(val -> " World");
}
public Observable<String> obString2() {
return Observable.just("")
.doOnNext(val -> {
System.out.println("Thread " + Thread.currentThread()
.getName());
})
.map(val -> "!");
}
public Observable<String> obAsyncString() {
return Observable.just("")
.observeOn(scheduler)
.doOnNext(val -> {
System.out.println("Thread " + Thread.currentThread()
.getName());
})
.map(val -> "Hello");
}
public Observable<String> obAsyncString1() {
return Observable.just("")
.observeOn(scheduler1)
.doOnNext(val -> {
System.out.println("Thread " + Thread.currentThread()
.getName());
})
.map(val -> " World");
}
public Observable<String> obAsyncString2() {
return Observable.just("")
.observeOn(scheduler2)
.doOnNext(val -> {
System.out.println("Thread " + Thread.currentThread()
.getName());
})
.map(val -> "!");
}
您可以在此处查看更多示例https://github.com/politrons/reactive
一个小例子：
val observableOne = Observable.just("Hello", "World")
val observableTwo = Observable.just("Bye", "Friends")
val zipper = BiFunction<String, String, String> { first, second -> "$first - $second" }
Observable.zip(observableOne, observableTwo, zipper)
.subscribe { println(it) }
这将打印：
Hello - Bye
World - Friends
在BiFunction<String, String, String>中第一个String第一个observable的类型，第二个String是第二个observable的类型，第三个String表示类型你的拉链函数的return。
我做了一个小例子，在 this blog post
中使用 zip 调用了两个真实端点我一直在寻找关于如何使用 Zip 运算符的简单答案，以及如何处理我创建的 Observables 以将它们传递给它，我想知道我是否应该为每个 observable 或不，这些答案都不是很容易找到的，我必须自己弄清楚，所以这是一个在 2 个 Observables 上使用 Zip 运算符的简单示例：
@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 指定这个 ...这是预期的行为）
这是我使用 Single.zip 和 rxJava2
的实现我尽量让它简单易懂
//
// API Client Interface
//
@GET(ServicesConstants.API_PREFIX + "questions/{id}/")
Single<Response<ResponseGeneric<List<ResponseQuestion>>>> getBaseQuestions(@Path("id") int personId);
@GET(ServicesConstants.API_PREFIX + "physician/{id}/")
Single<Response<ResponseGeneric<List<ResponsePhysician>>>> getPhysicianInfo(@Path("id") int personId);
//
// API middle layer - NOTE: I had feedback that the Single.create is not needed (but I haven't yet spent the time to improve it)
//
public Single<List<ResponsePhysician>> getPhysicianInfo(int personId) {
return Single.create(subscriber -> {
apiClient.getPhysicianInfo(appId)
.subscribeOn(Schedulers.io())
.observeOn(Schedulers.io())
.subscribe(response -> {
ResponseGeneric<List<ResponsePhysician>> responseBody = response.body();
if(responseBody != null && responseBody.statusCode == 1) {
if (!subscriber.isDisposed()) subscriber.onSuccess(responseBody.data);
} else if(response.body() != null && response.body().status != null ){
if (!subscriber.isDisposed()) subscriber.onError(new Throwable(response.body().status));
} else {
if (!subscriber.isDisposed()) subscriber.onError(new Throwable(response.message()));
}
}, throwable -> {
throwable.printStackTrace();
if(!subscriber.isDisposed()) subscriber.onError(throwable);
});
});
}
public Single<List<ResponseQuestion>> getHealthQuestions(int personId){
return Single.create(subscriber -> {
apiClient.getBaseQuestions(personId)
.subscribeOn(Schedulers.io())
.observeOn(Schedulers.io())
.subscribe(response -> {
ResponseGeneric<List<ResponseQuestion>> responseBody = response.body();
if(responseBody != null && responseBody.data != null) {
if (!subscriber.isDisposed()) subscriber.onSuccess(response.body().data);
} else if(response.body() != null && response.body().status != null ){
if (!subscriber.isDisposed()) subscriber.onError(new Throwable(response.body().status));
} else {
if (!subscriber.isDisposed()) subscriber.onError(new Throwable(response.message()));
}
}, throwable -> {
throwable.printStackTrace();
if(!subscriber.isDisposed()) subscriber.onError(throwable);
});
});
}
//please note that ResponseGeneric is just an outer wrapper of the returned data - common to all API's in this project
public class ResponseGeneric<T> {
@SerializedName("Status")
public String status;
@SerializedName("StatusCode")
public float statusCode;
@SerializedName("Data")
public T data;
}
//
// API end-use layer - this gets close to the UI so notice the oberver is set for main thread
//
private static class MergedResponse{// this is just a POJO to store all the responses in one object
public List<ResponseQuestion> listQuestions;
public List<ResponsePhysician> listPhysicians;
public MergedResponse(List<ResponseQuestion> listQuestions, List<ResponsePhysician> listPhysicians){
this.listQuestions = listQuestions;
this.listPhysicians = listPhysicians;
}
}
// example of Single.zip() - calls getHealthQuestions() and getPhysicianInfo() from API Middle Layer
private void downloadHealthQuestions(int personId) {
addRxSubscription(Single
.zip(getHealthQuestions(personId), getPhysicianInfo(personId), MergedResponse::new)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(response -> {
if(response != null) {
Timber.i(" - total health questions downloaded %d", response.listQuestions.size());
Timber.i(" - physicians downloaded %d", response.listPhysicians.size());
if (response.listPhysicians != null && response.listPhysicians.size()>0) {
// do your stuff to process response data
}
if (response.listQuestions != null && response.listQuestions.size()>0) {
// do your stuff to process response data
}
} else {
// process error - show message
}
}, error -> {
// process error - show network error message
}));
}
您将 rxjava 中的 zip 与 Java 8 一起使用：
Observable<MovieResponse> movies = ...
Observable<PictureResponse> picture = ...
Observable<ZipResponse> response = Observable.zip(movies, picture, ZipResponse::new);
class ZipResponse {
private MovieResponse movieResponse;
private PictureResponse pictureResponse;
ZipResponse(MovieResponse movieResponse, PictureResponse pictureResponse) {
this.movieResponse = movieResponse;
this.pictureResponse = pictureResponse;
}
public MovieResponse getMovieResponse() {
return movieResponse;
}
public void setMovieResponse(MovieResponse movieResponse) {
this.movieResponse= movieResponse;
}
public PictureResponse getPictureResponse() {
return pictureResponse;
}
public void setPictureResponse(PictureResponse pictureResponse) {
this.pictureResponse= pictureResponse;
}
}