android - RxJava - 并行执行两个 API 调用

Question

我有一个 SyncService。

@Override
public int onStartCommand(Intent intent, int flags, final int startId) {
    Timber.i("Starting sync...");
    ...
    RxUtil.unsubscribe(mSubscription);
    mSubscription = mDataManager.syncEvents()
            .subscribeOn(Schedulers.io())
            .subscribe(new Observer<Event>() {
                @Override
                public void onCompleted() {
                    Timber.i("Synced successfully!");
                    stopSelf(startId);
                }

                @Override
                public void onError(Throwable e) {
                    Timber.w(e, "Error syncing.");
                    stopSelf(startId);

                }

                @Override
                public void onNext(Event event) {
                }
            });

    return START_STICKY;
}

Observable events = mDataManager.syncEvents() 是一个 API 调用。

我想做一个并行调用:

单用户信息 = mDataManager.getUserInfo()

并调用 stopSelf(startId);完成这两个调用后。

我该怎么做？

我试过了 RxJava Fetching Observables In Parallel但这有点不同。

我想我必须使用 .zip 或 .merge 方法。但在我的例子中，一个方法调用返回 Observable(事件列表)和第二个 Single(一个 UserInfo 对象)。

我创建了 z 结果类，它可能是 .zip 方法的结果，但我不知道如何填充它:

public class SyncResponse {
     List<Event> events;
     UserInfo userInfo;
     ...
}

Answer 1

由于您只有两个可观察对象，您可以使用 android Pair 将它们组合起来以组合结果。

mDataManager.syncEvents()
    .zipWith(mDataManager.getUserInfo().toObservable(), Pair::create)
    .subscribeOn(Schedulers.io())
    .subscribe(pair -> {
        Event event = pair.first;
        UserInfo userInfo = pair.second;

        // your code here

        Timber.i("Synced successfully!");
        stopSelf(startId);
    }, throwable -> {
        Timber.w(e, "Error syncing.");
        stopSelf(startId);
    });

你可以在没有 java8/retrolmbda 的情况下做同样的事情，但是为什么

如果您需要收集所有事件直到完成并将它们全部与单个用户信息结合起来，这会有点复杂

Observable<ArrayList<String>> eventsObservable = mDataManager.syncEvents()
    .collect(ArrayList::new, ArrayList::add); 

mDataManager.getUserInfo()
    .zipWith(eventsObservable.toSingle(), Pair::create)
    .subscribeOn(Schedulers.io())
    .subscribe(pair -> {
        UserInfo userInfo = pair.first;
        List<Event> event = pair.second;

        // your code here

        Timber.i("Synced successfully!");
        stopSelf(startId);
    }, throwable -> {
        Timber.w(e, "Error syncing.");
        stopSelf(startId);
    });