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

【问题标题】：Zip operator in RxJava is not working with RetrofitRxJava 中的 Zip 运算符不适用于 Retrofit
【发布时间】：2017-07-06 07:20:44
【问题描述】：

我正在尝试在 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());
        /**
        ***
        */
    }
});

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

我的依赖项有问题吗？
应该使用 AndroidSchedulers.mainThread() 代替 Schedulers.io()

【问题讨论】：

看起来你的一个（或多个）改造 Observable 要么发出错误，要么根本没有发出任何东西。您可能希望在您的 onError 和 onCompleted 中添加一些日志记录。
是的，我尝试过这样做，但 URL 的格式存在一些问题，因此抛出了 404 异常。我调试并发现了这个问题。感谢您的帮助！

标签： java android observable retrofit2 rx-java2

【解决方案1】：

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

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

干杯！

【讨论】：