rx-java：订阅IntentService的onHandleIntent中的Observable

Question

我遇到了以下无法解决的问题：

我有一个 IntentService，它在 onHandleIntent 方法中执行一些工作。除了此方法中的一些常见内容外，我还必须向后端发出请求并发送所需的信息。众所周知，onHandleIntent 方法在自己的线程中工作，并在执行最后一条命令的那一刻停止。在某些异步操作的情况下 - 结果可能会丢失，因此 google 建议仅在此方法中使用同步操作。

这是我的问题，我没有使用 subscribeOn 和 observeOn，但看起来我的方法在 sep 有效。线程，这对我来说不安全。

这是我的 onHandleIntent 方法的主体，带有 Retrofit（带有 rx.Observable）调用：

Log.d(TAG, "onHandleIntent | started");

...

getAPI().addItem(item).timeout(AppConfig.NETWORK_TIMEOUT, TimeUnit.SECONDS).doOnNext(new Action1<Response<String>>() {
        @Override
        public void call(Response<String> response) {
           //some job
          Log.d(TAG, "doOnNext || call");
        }
    }).retryWhen(new RetryWithSessionRefresh()).subscribe(new Observer<Response<String>>() {
            @Override
            public void onCompleted() {
                Log.d(TAG, "Observer onCompleted");
            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "Observer onError");
            }

            @Override
            public void onNext(Response<String> response) {
                Log.d(TAG, "Observer onNext");
            }
        });

...

Log.d(TAG, "onHandleIntent | finished");

改造方法getItem(item)为：

 rx.Observable<Response<String>> addItem(@Body ArrayList<Item> item);

最后的结果是：

• 11-20 00:49:03.512 onHandleIntent |开始
• 11-20 00:49:03.517 onHandleIntent |完成
• 11-20 00:49:04.549 doOnNext ||打电话
• 11-20 00:49:04.549 观察者 onNext
• 11-20 00:49:04.550 观察者已完成

那么，问题是如何让它在同一个线程中工作？

Answer 1

您始终可以将 RestAdapter 实例使用的 Http Executer 更改为同步的。 Retrofit 在内部使用 Http Executer 来调度Observable。这将允许您在订阅的线程上启动流程。这也意味着你不能使用这个RestAdapter，如果你不打算像 IntentService 那样提供你自己的Thread。

Executor immediateExecuter = new Executor() {
            @Override
            public void execute(Runnable command) {
                command.run();
            }
        };

... new RestAdapter.Builder().setExecutors(immediateExecuter, null).build();

我认为你也可以利用 RxJava 提供的工具来解决这个问题。您可以使用toBlocking 将您的Observable 变成BlockingObservable。这将允许您在继续执行代码之前等待您的流完全完成。需要考虑的一件事是，您必须使用 try/catch 块包装整个流程来处理异常，因为现在将抛出流程中未捕获的异常。我想你可以说这会将你的流变成一个重要的命令式语句。

try {
    getAPI().addItem(item).timeout(AppConfig.NETWORK_TIMEOUT, TimeUnit.SECONDS).doOnNext(new Action1<Response<String>>() {
            @Override
            public void call(Response<String> response) {
               //some job
              Log.d(TAG, "doOnNext || call");
            }
        })
        .retryWhen(new RetryWithSessionRefresh())
        .toBlocking()
        .forEach(new Action1<Response<String>>() {
                @Override
                public void call(Response<String> s) {
                    Log.d(TAG, "Observer onNext");
                }
        });
} catch (WhatEverExceptionsYourStreamMightThrow e) {
  // Handle error here
}

...

Log.d(TAG, "onHandleIntent | finished");

我认为重要的是要注意，每次运行 IntentService 时，第二种解决方案总是涉及 2 个线程。一个线程将由 Retrofit 使用，另一个将是 IntentService 本身。