RxJava - 如何在订阅的 lambda 函数中启动异步网络调用？可观察的

Question

以下是我的确切代码。请注意，在 subscribe lambda 函数中调用了 foo(o)。这会在 Android 中产生严格模式错误。现在我当然可以在这里分拆一个线程，但那是 Rxjava 的做事方式吗？我现在是否必须再次考虑线程？想知道我在这里的最佳选择。我怎样才能做到这种 RxJava 方式，以便代码的其他部分可以订阅将返回网络调用响应的新事件？所以本质上我是通过异步响应将一种类型的事件 MyObj 转换为 NetObj 响应事件。

 public static final PublishSubject<MyObj> myObjSubject = PublishSubject.create();
  public static final Observable<MyObj> observable = myObjSubject.asObservable();


protected CompositeSubscription myCompositeSubs = new CompositeSubscription();

myCompositeSubs.add(
            observable.subscribe((MyObj o) -> {
                 // `. Update UI with MyObject and 
                 // 2. Kick off network call which will return JsonResponse type object which I would like to possibly process here and also publish to let others respond to it.
                 foo(o);  // $$$$$ million dollar question what if I am kicking off something long running here?
            }));


private void foo(MyObj o){

// long running request.  How to kick off long running from here? 
  JsonRespons resp = RestAdapter.makeQuest(o.url);
  // I want this JsonResponse object to be an even others can listen for. I don't want to just process it right here.

}

更新：只是为了澄清我想要做的是在类（Android 片段）中侦听数据事件 MyObject。一旦我得到这个，我想做两件事，用 MyObject 同步更新 UI，这是我想要消费的东西，而不仅仅是转换。但我也想启动一个长期运行的请求。返回 JsonObject 响应的网络调用。当它返回时，我可能会在同一个片段中处理它，或者我可能希望其他类能够监听这个事件。所以我不只是观察 MyObject 然后进行网络调用并获取响应。我需要同时执行这两项操作，一项同步另一项异步。

Answer 1

这是使用像flatMap 这样的 RxJava 运算符的好机会。如果您需要更多文档，请查看 here，但 flatMap 允许您获取一个类型的 Observable（您的 Observable<MyObj>）并返回另一种类型的 Observable（在这种情况下，它可以是 Observable<Void> 或者可能Observable<Boolean> 如果您想知道您的长期运行请求是否完成）。

看看这段代码：

Subscription sub = observable.flatMap((MyObj o) -> {
      // this will be run on your subscribeOn thread -- in background
      return Observable.just(foo(o));
    }).subscribe(aVoid -> {
      // we have completed both operations here, emissions will be 
      // on our observeOn() thread
    });

foo(MyObj) 函数现在的结构如下：

private Void foo(MyObj o) {
    // long running stuff
    return null;
}

将返回类型设为Void 而不是void 很重要，这样我们才能使用flatMap 运算符并返回Observable<Void>。

您所有长时间运行的foo() 操作将在您的subscribeOn() 线程上运行，以及您的初始可观察对象，一旦操作完成，它将在您的@ 上向您的subscribe() 发出一个Void 项目987654338@ 线程（如果您想通知 UI，这可以是您在 Android 中的主线程）。

已更新以允许 UI 在 flatMap 调用中工作：

注意：这假定第一个 observable 已在后台线程上定义了 subscribeOn() 并且将 observeOn() 定义为 AndroidSchedulers.mainThread()

Subscription sub = observable.flatMap((MyObj o) -> {
      // << do stuff with MyObj to UI HERE >> 
      return Observable.defer(() -> Observable.just(foo(o)))
          .subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread());
    }).subscribe(aVoid -> {
      // foo() has completed, we're back on the UI thread
    });

如果您想使用 JsonResponse 做某事，您可以更改 foo() 函数以返回它而不是 Void，然后您将能够在 UI 线程上使用 foo() 的结果。

更新 2： 将 Observable.create() 更改为 Observable.defer() 以使其更干净。