正确使用Retrofit + RxJava的combineLatest

Question

我想异步执行2个网络调用 - 我正在使用Retrofit + RxJava来完成此任务.这个逻辑来自一个简单的Runner类来测试解决方案.注意:这主要涉及服务器端的RxJava.

我的代码如下所示:

public static void main(String[] args) throws Exception {
  Api api = ...;

  Observable.combineLatest(
      api.getStates(),
      api.getCmsContent(),
      new Func2<List<States>, CmsContent, String>() {
        @Override public String call(List<State> states, CmsContent content) {
          ...
          return "PLACEHOLDER";
        }
      })
      .observeOn(Schedulers.immediate())
      .subscribeOn(Schedulers.immediate())
      .subscribe(new Observer<String>() {
        @Override public void onCompleted() {
          System.out.println("COMPLETED");
        }

        @Override public void onError(Throwable e) {
          System.out.println("ERROR: " + e.getMessage());
        }

        @Override public void onNext(String s) {
          // I don't care what's returned here
        }
      });
}

三个问题:

1. Observable.combineLatest当您想要异步执行多个REST调用并在所有调用完成后继续操作时,是最好的运算符吗？
2. 我的Func2实现目前返回一个String.执行2个API调用后,我将在Func2#call()方法中处理结果 .我不在乎返回什么 - 必须有一个更好的方法来处理这个问题 - 我是否正确？
3. 使用上面的代码正确执行API调用.但是当我运行程序时,该main方法没有正确完成Process finished with exit code 0.可能导致代码挂起的原因是什么？

更新 - 2015-05-14

根据建议,我已将逻辑更改为以下内容:

public static void main(String[] args) throws Exception {
  Api api = ...;

  Observable.zip(
      api.getStates(),
      api.getCmsContent(),
      new Func2<List<States>, CmsContent, Boolean>() {
        @Override public Boolean call(List<State> states, CmsContent content) {
          // process data
          return true;
        }
      })
      .subscribeOn(Schedulers.io())
      .toBlocking()
      .first();
}

这看起来像我正在寻找的解决方案.我将使用它一段时间,看看我是否遇到任何麻烦.

Answer 1

1)如果你知道你在两条路径上都有一个值,那就好了zip.

2)你想做什么？你会得到你的价值观Func2,如果你真的不关心旅行onNext的价值,那就回复你的选择.

3)Schedulers.immediate()在某种意义上不是真正的调度程序,并且很容易出现相同的池死锁情况.你真的不需要使用它.如果要在异步工作完成之前阻止主线程,请使用toBlocking().first()例如.

Answer 2

1）不，最好是使用zip()。如果两个（或多个）api 之一返回“较慢”的不同结果/它具有缓存的本质，则合并最新的结果是很好的。

2）Fu​​n2有利于合并结果。最好（从架构角度）在 onNext() 或 onError() 中处理结果。您可以使用一个简单的Pair<T,Y>类将结果从 Func2 传递到 onNext()。

3）没有什么问题。上述结果应该在 onNext() 中处理，而不是在 onComplete 中处理。根据Retrofit的源代码，结果仅在onNext()中传递（当然是正确的）。

希望那些有帮助。