RxAndroid Observable 跳过 doOnComplete()

Question

我同时调用多个 Retrofit api 并等待其完成，这由 Observable.zip() 正确完成。现在我想实现进步。

这是我的实现..

private void preFetchData() {
    ApiInterface apiService1 = ApiClient.getWooRxClient().create(ApiInterface.class);
    ApiInterface apiService2 = ApiClient.getRxClient().create(ApiInterface.class);
    Map<String, String> map1 = new HashMap<>();
    map1.put("on_sale", "true");
    Observable<List<Product>> call1 = apiService1.getProducts1(map1)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .doOnComplete(() -> {
                progress += 18;
                Log.e("Progress1", progress + "");
                mProgressBar.setProgress(progress);
                mProgressBar.setProgressText(progress + "%");
            });

    Map<String, String> map2 = new HashMap<>();
    map2.put("featured", "true");
    Observable<List<Product>> call2 = apiService1.getProducts1(map2)
            .subscribeOn(Schedulers.io())
            .delaySubscription(100, TimeUnit.MILLISECONDS)
            .observeOn(AndroidSchedulers.mainThread())
            .doOnComplete(() -> {
                progress += 18;
                Log.e("Progress2", progress + "");
                mProgressBar.setProgress(progress);
                mProgressBar.setProgressText(progress + "%");
            });

    Map<String, String> map3 = new HashMap<>();
    map3.put("page", "1");
    map3.put("sort", "rating");
    map3.put("per_page", "10");
    Observable<List<Product>> call3 = apiService2.getCustomProducts1(map3)
            .subscribeOn(Schedulers.io())
            .delaySubscription(200, TimeUnit.MILLISECONDS)
            .observeOn(AndroidSchedulers.mainThread())
            .doOnComplete(() -> {
                progress += 18;
                Log.e("Progress3", progress + "");
                mProgressBar.setProgress(progress);
                mProgressBar.setProgressText(progress + "%");
            });

    Map<String, String> map4 = new HashMap<>();
    map4.put("page", "1");
    map4.put("sort", "popularity");
    map4.put("per_page", "10");
    Observable<List<Product>> call4 = apiService2.getCustomProducts1(map4)
            .subscribeOn(Schedulers.io())
            .delaySubscription(300, TimeUnit.MILLISECONDS)
            .observeOn(AndroidSchedulers.mainThread())
            .doOnComplete(() -> {
                progress += 18;
                Log.e("Progress4", progress + "");
                mProgressBar.setProgress(progress);
                mProgressBar.setProgressText(progress + "%");
            });

    Observable<ResponseBody> call5 = apiService2.getCurrencySymbol()
            .subscribeOn(Schedulers.io())
            .delaySubscription(400, TimeUnit.MILLISECONDS)
            .observeOn(AndroidSchedulers.mainThread())
            .doOnComplete(() -> {
                progress += 10;
                Log.e("Progress5", progress + "");
                mProgressBar.setProgress(progress);
                mProgressBar.setProgressText(progress + "%");
            });

    Observable<List<Category>> call6 = apiService1.getAllCategories()
            .subscribeOn(Schedulers.io())
            .delaySubscription(500, TimeUnit.MILLISECONDS)
            .observeOn(AndroidSchedulers.mainThread())
            .doOnComplete(() -> {
                progress += 18;
                Log.e("Progress6", progress + "");
                mProgressBar.setProgress(progress);
                mProgressBar.setProgressText(progress + "%");

            });

    Observable<CombinedHomePage> combined = Observable.zip(call1, call2, call3, call4, call5, call6, CombinedHomePage::new);
    disposable = combined.subscribe(this::successHomePage, this::throwableError);
}

private void successHomePage(CombinedHomePage o) {
    Log.e("Response", "SUCCESS " + o.featuredProductList.size());
    Log.e("Response", "SUCCESS " + o.saleProductList.size());
    Log.e("Response", "SUCCESS " + o.topRatedProductList.size());
    Log.e("Response", "SUCCESS " + o.topSellerProductList.size());
    Log.e("Response", "SUCCESS " + o.CURRENCY);
    Log.e("Response", "SUCCESS " + o.categoryList.size());
}

private void throwableError(Throwable t) {
    Log.e("Response", "Fail");
}

这是 Logcat

第一次运行

E/Progress5: 10.0
E/Progress2: 28.0
E/Progress1: 46.0
E/Progress6: 64.0
E/Progress3: 82.0
E/Response: Featured List Size 5
E/Response: Sale List Size 7
E/Response: Rated List Size 10
E/Response: Seller List Size 10
E/Response: Currency $
E/Response: Category List Size 9

第二次运行

E/Progress5: 10.0
E/Progress2: 28.0
E/Progress1: 46.0
E/Progress6: 64.0
E/Progress4: 82.0
E/Response: Featured List Size 5
E/Response: Sale List Size 7
E/Response: Rated List Size 10
E/Response: Seller List Size 10
E/Response: Currency $
E/Response: Category List Size 9

第一次运行 Progess 4 跳过，第二次运行 Progess 3 跳过（所有 Api 均已正确完成）。

知道为什么会这样吗？

如有任何帮助，将不胜感激

Answer 1

在javadoc中有描述：

如果其中一个源比其他源短，则操作员按指定的顺序订阅其源，并在处置其他源时急切地完成。因此，那些其他源可能永远无法运行完成（因此不会调用 doOnComplete()）。如果源的长度完全相同，也会发生这种情况；如果源 A 完成且 B 已被消耗且即将完成，则操作员检测到 A 不会发送进一步的值，它将立即处理 B。例如：

zip(Arrays.asList(range(1, 5).doOnComplete(action1), range(6, 5).doOnComplete(action2)), (a) -> a)

action1 将被调用，但 action2 不会。 要解决此终止属性，请同时使用 doOnDispose(Action) 或使用 using() 在完成或 dispose() 调用的情况下进行清理。