android - 如何使用 RxJava2 使用 IntentService 线程

Question

我面临下一个情况。我在 onHadleIntent 方法中开始我的代码，部分代码在 IntentService 线程中工作，但 getInfoAboutUser() 中的 Observable.zip 方法在 RxJava 线程中工作。

@Override
protected void onHandleIntent(@Nullable Intent intent) {
      LOG.debug(Thread.currentThread().getName());

      Call<String> call= mRepository.getInfo();
        try {
            retrofit2.Response<String> response = call.execute();
            if (response.isSuccessful()) {
                LOG.debug("Response body "+Thread.currentThread().getName());
                getInfoAboutUser();
            }
       }catch(){}
}

public void getInfoAboutUser(){
       LOG.debug("getInfoAboutUser "+Thread.currentThread().getName());

       Executor e = new Executor() {
            @Override
            public void execute(@NonNull Runnable runnable) {
                LOG.debug(" executor thread");
                runnable.run();
            }
        }; 

 Observable.zip(
                Observable.fromIterable(array),
                Observable
                        .interval((mRandom.nextInt(7)+5) * 1000, 
 TimeUnit.MILLISECONDS,Schedulers.from(e))
                        .take(array.size()),
                new BiFunction<String, Long, String>() {
                    @Override
                    public String apply(String s, Long aLong) throws Exception {
                        LOG.debug("Result "+Thread.currentThread().getName());
                        return s;
                    }
                }
       ).flatMapMaybe(new Function<String, MaybeSource<String>>() {
            @Override
            public MaybeSource<String> apply(String s) throws Exception {
                return mRepository.getInfoAboutUser(s); 
            }
        }).subscribeWith(new DisposableObserver<String>() {})
}

我使用的 mRepository.getInfo() 和 mRepository.getInfoAboutUser(s) 方法没有 subscribeOn 和 observeOn!

我的日志是:

• IntentService 线程
• IntentService 线程响应体
• IntentService 线程 getInfoAboutUser
• RxSingleScheduler-1 执行线程
• RxSingleScheduler-1 结果

等等

如何将 IntentService 线程用于 Observable.zip 和 Interval 方法？我只需要 IntentService 线程

Answer 1

Schedulers.from(e) 将 Executor 包装在内部类 ExecutorScheduler 中。如果任务被延迟安排并且给定的 Executor 不是 ScheduledExecutorService 那么它将使用内部调度程序来延迟对 e.execute() 的调用 直到延迟结束。

因为您的执行器只是立即执行，所以它最终会在 RxSingleScheduler-1 辅助调度器上执行。

要解决此问题，您需要从以下解决方案中进行选择:

1. 创建一个 ScheduledExecutorService，它将 runnable 正确地分派(dispatch)给 IntentService。
2. 创建一个自定义 Scheduler，它将 runnable 分派(dispatch)给 IntentService 的 Looper。
3. 使用RxAndroid图书馆为你做 2. AndroidSchedulers.from(Looper.myLooper()) 将为 IntentService 创建一个调度程序。

编辑:请注意，IntentService 和异步操作不能混用。当 handleIntent 返回时，服务将终止，所以这不是一个好主意 执行延迟操作的方法，例如 Observable.interval。