RxJava 中的 Zip 运算符不适用于 Retrofit

Question

我正在尝试在 android 中的 RxJava 中使用 zip 运算符，我正在尝试执行 3 个并行 API 调用以将它们的结果放在一起。但是我的 zip 运算符没有产生结果。我的示例问题的代码如下：

我的 gradle 文件的代码

compile 'com.squareup.retrofit2:retrofit:2.0.2'
compile 'com.squareup.retrofit2:converter-gson:2.0.2'
compile 'com.squareup.retrofit2:adapter-rxjava:2.0.2'

我也将其包含在我的 gradle 文件中

exclude 'META-INF/rxjava.properties'

我改造客户端的代码

retrofit = new Retrofit.Builder().baseUrl(BASE_URL)
    .addCallAdapterFactory(RxJavaCallAdapterFactory.create())
    .addConverterFactory(GsonConverterFactory.create())
    .client(client).build();

我的 API 界面的代码

public interface ApiInterface {
    @GET("/users/{UUID}/count.json")
    Observable<Count> getCountInfo(@Path("UUID") String UUID, @Query("store_id") String sort);
    @GET("v1/users/{UUID}.json")
    Observable<GetStatus> getState(@Path("UUID") String UUID);
    @GET("v1/user/{UUID}/points.json")
    Observable<Response> getResponse(@Path("UUID") String UUID);
}

我的 Observables 代码是

Retrofit repo = APIClient.getClient(baseUrl);
Observable<Count> userObservable = repo.create(ApiInterface.class)
    .getCount(userid,"1")
    .subscribeOn(Schedulers.newThread())
    .observeOn(Schedulers.io());
Observable<GetStatus> eventObservable = APIClient.getClient(baseUrl)
    .create(ApiInterface.class)
    .getState(userid)
    .subscribeOn(Schedulers.newThread())
    .observeOn(Schedulers.io());
Observable<Response> eventsObservable1 = APIClient
    .getClient(baseUrl)
    .create(ApiInterface.class)
    .getPoints(userid)
    .subscribeOn(Schedulers.newThread())
    .observeOn(Schedulers.io());

我的组合 observable 和 zip 运算符的代码是：

Observable<CommonSplashResponse> combined = Observable.zip(userObservable, eventsObservable, eventsObservable1,
new Func3<Count, GetStatus, Response, CommonResponse>() {
    @Override
    public CommonResponse call(Count count, GetStatus uStatus, 
        Response lResponse) {
        return new CommonResponse(count, uStatus, lResponse);
    }
});
combined.subscribe(new Subscriber<CommonSplashResponse>() {
    @Override
    public void onCompleted() {
    }
    @Override
    public void onError(Throwable e) {
    }
    @Override
    public void onNext(CommonResponse o) {
        LOG.info("Count Value is " + o.getCount());
        /**
        ***
        */
    }
});

我面临的问题是，Combined Observable 的 onNext 中的语句没有被执行。执行中出现故障的原因可能是什么？所以我想问一下：

我的依赖有问题吗？
应该使用 AndroidSchedulers.mainThread() 而不是 Schedulers.io()

Answer 1

要让 .zip() 运算符发射任何东西，所有压缩的可观察对象都必须至少发射一次。如果您的一个可观察对象发出错误，或根本不发出，您将永远不会收到 onNext 事件。

为了检查错误排放，将日志记录或断点添加到订阅中的 onError
为了检查丢失的发射，您可以在所有压缩的 Observable 之后添加带有日志记录的 doOnNext 和 doOnCompleted 调用，看看哪个没有发射

干杯！

RxJava 中的 Zip 运算符不适用于 Retrofit

Zip operator in RxJava is not working with Retrofit

java

android

observable

retrofit2

rx-java2