等待两个 observables 完成

Question

我有两个列表 - 每个列表都有它的 API，对于每个项目我都必须调用 API（使用 retrofit2 - android）。

我可以这样做吗？

我如何订阅他们的两个完成？

    Observable<Integer> offerObservable = Observable.from(restDatabaseImages.getOfferImages());
    Observable<Integer> otherOfferObservable = Observable.from(restDatabaseImages.getOtherOfferImages());

    offerObservable.flatMap(new Func1<Integer, Observable<RestOfferImage>>() {
        @Override
        public Observable<RestOfferImage> call(Integer integer) {
            return backendService.getDatabaseOfferImage(integer);
        }
    }).subscribe(new Action1<RestOfferImage>() {
        @Override
        public void call(RestOfferImage restOfferImage) {
            offerService.saveOfferImage(restOfferImage);
        }
    });

    otherOfferObservable.flatMap(new Func1<Integer, Observable<RestOtherOfferImage>>() {
        @Override
        public Observable<RestOtherOfferImage> call(Integer integer) {
            return backendService.getDatabaseOtherOfferImage(integer);
        }
    }).subscribe(new Action1<RestOtherOfferImage>() {
        @Override
        public void call(RestOtherOfferImage restOtherOfferImage) {
            otherOfferService.saveOtherOfferImage(restOtherOfferImage);
        }
    });

-- 更新：

我试着把它改写成这样。问题是我只看到执行了第一个 flatMap Func1 - 对于第一个条目...

Observable<Integer> offerObservable = Observable.from(restDatabaseImages.getOfferImages());
Observable<Integer> otherOfferObservable = Observable.from(restDatabaseImages.getOtherOfferImages());

Observable ob1 = offerObservable.flatMap(new Func1<Integer, Observable<RestOfferImage>>() {
    @Override
    public Observable<RestOfferImage> call(Integer integer) {
        return backendService.getDatabaseOfferImage(integer);
    }
}).map(new Func1<RestOfferImage, Void>() {
    @Override
    public Void call(RestOfferImage restOfferImage) {
        offerService.saveOfferImage(restOfferImage);
        return null;
    }
});

Observable ob2 = otherOfferObservable.flatMap(new Func1<Integer, Observable<RestOtherOfferImage>>() {
    @Override
    public Observable<RestOtherOfferImage> call(Integer integer) {
        return backendService.getDatabaseOtherOfferImage(integer);
    }
}).map(new Func1<RestOtherOfferImage, Void>() {
    @Override
    public Void call(RestOtherOfferImage restOtherOfferImage) {
        otherOfferService.saveOtherOfferImage(restOtherOfferImage);
        return null;
    }
});

ob1.zipWith(ob2, new Func2() {
    @Override
    public Object call(Object o, Object o2) {
        return null;
    }
}).subscribe(new Subscriber() {
    @Override
    public void onCompleted() {
        Log.d("AllDone", "DA");
        onLoadingFinishedListenerCallback.onLoading2Finished();
    }

    @Override
    public void onError(Throwable e) {

    }

    @Override
    public void onNext(Object o) {

    }
});

Answer 1

我认为您正在寻找这样的东西：

public class SimpleTest {

    @Test
    public void testSample() {
        Integer[] offerImages = {1, 2, 3, 4, 5};
        Integer[] otherOfferImages = {6, 7, 8, 9, 10};

        Observable<Integer> offerObservable = Observable.from(offerImages).flatMap(someInt -> fakeApiCall(someInt));
        Observable<Integer> otherOfferObservable = Observable.from(otherOfferImages).flatMap(someInt -> fakeApiCall(someInt));

        Observable.zip(
                offerObservable,
                otherOfferObservable,
                (offer, otherOffer) -> String.format("Offer: %d - Other Offer: %d", offer, otherOffer)
        ).subscribe(data -> System.out.println(data));
    }

    private Observable<? extends Integer> fakeApiCall(Integer someInt) {
        return Observable.just(someInt * 10);
    }
}

Zip 运算符将确保您将这些事件一起发出，然后您可以使用它们生成一个新流。

上面的代码会打印这个：

Offer: 10 - Other Offer: 60
Offer: 20 - Other Offer: 70
Offer: 30 - Other Offer: 80
Offer: 40 - Other Offer: 90
Offer: 50 - Other Offer: 100