RxJava：如何从上游保留线程

Question

在Android上，总是通过设计在主Thread上调用一些回调。以下ServiceConnection：

private Completable dismissService() { 
    return Completable.fromEmitter(new Action1<CompletableEmitter>() { 
     @Override 
     public void call(final CompletableEmitter completableEmitter) { 
      final ServiceConnection conn = new ServiceConnection() { 
       @Override 
       public void onServiceConnected(ComponentName name, IBinder service) { 
        // here always main Thread... 
        unbindService(this); 
        completableEmitter.onCompleted(); 
       } 

       @Override 
       public void onServiceDisconnected(ComponentName name) { 
        // no-op 
       } 
      }; 

      completableEmitter.setCancellation(new AsyncEmitter.Cancellable() { 
       @Override 
       public void cancel() throws Exception { 
        unbindService(conn); 
       } 
      }); 

      bindService(new Intent(MainActivity.this, MyService.class), conn, BIND_AUTO_CREATE); 
     } 
    }); 
} 

不过，我想通过Completable返回dismissService()发射其结果就什么Thread它被调用的。我试图用一个newSingleThreadExecutor()以下（哈克？）解决方案：

private Completable dismissServiceRetainingThread() { 
    return Single.fromCallable(new Callable<Thread>() { 
     @Override 
     public Thread call() throws Exception { 
      return Thread.currentThread(); 
     } 
    }).flatMapCompletable(new Func1<Thread, Completable>() { 
     @Override 
     public Completable call(final Thread thread) { 
      return Completable.fromEmitter(new Action1<CompletableEmitter>() { 
       @Override 
       public void call(final CompletableEmitter completableEmitter) { 
        final ServiceConnection conn = new ServiceConnection() { 
         @Override 
         public void onServiceConnected(ComponentName name, IBinder service) { 
          // here always main Thread... 
          unbindService(this); 
          completableEmitter.onCompleted(); 
         } 

         @Override 
         public void onServiceDisconnected(ComponentName name) { 
          // no-op 
         } 
        }; 

        completableEmitter.setCancellation(new AsyncEmitter.Cancellable() { 
         @Override 
         public void cancel() throws Exception { 
          unbindService(conn); 
         } 
        }); 

        bindService(new Intent(MainActivity.this, MyService.class), conn, BIND_AUTO_CREATE); 
       } 
      }).observeOn(Schedulers.from(
          Executors.newSingleThreadExecutor(
          new ThreadFactory() { 
           @Override 
           public Thread newThread(@NonNull Runnable runnable) { 
           return thread; 
           } 
          } 
      ))); 
     } 
    }); 
} 

但是，有以下IllegalThreadStateException崩溃：

E/AndroidRuntime: FATAL EXCEPTION: main 
       Process: com.jenzz.rxjavathreadingtest, PID: 5307 
       java.lang.IllegalThreadStateException 
        at java.util.concurrent.ThreadPoolExecutor.addWorker(ThreadPoolExecutor.java:930) 
        at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1348) 
        at java.util.concurrent.Executors$DelegatedExecutorService.execute(Executors.java:591) 
        at rx.internal.schedulers.ExecutorScheduler$ExecutorSchedulerWorker.schedule(ExecutorScheduler.java:79) 
        at rx.Completable$28$1.onCompleted(Completable.java:1805) 
        at rx.internal.operators.CompletableFromEmitter$FromEmitter.onCompleted(CompletableFromEmitter.java:73) 
        at com.jenzz.rxjavathreadingtest.MainActivity$5$2$1.onServiceConnected(MainActivity.java:117) 

任何想法如何，我可以跳回原始Thread是以前用于上游，例如使用subscribeOn(Schedulers.io())？

Answer 1

你不想回到你所在的确切线程。当同一Scheduler中的其他线程可用时，它可能正忙，除非它是主线程，否则无法保证回调返回时它甚至仍然存在。只要您的call方法返回，它可用于更多工作或清理。我想你想要的是在原来的电话上执行相同的Scheduler。 AFIAK，没有直接的方法来确定线程的当前Scheduler（有一个脆弱的方式来确定一些调度，在这篇文章的底部提到）。所以你不能轻易做你想做的事情。默认通知主线程似乎是这种方法的理智行为。如果您希望它默认为不同的Scheduler，则可以使用.observeOn(Schedulers.io())或您选择的计划程序。

正如其他答案提到的，你可以尝试创建一个Handler如果当前线程都有一个活套。回调发生时，您仍然依靠调用者来确保线程仍然可行。这似乎是一个更高级别的责任，而不是告诉来电者的回应将出现在主线程中，除非他们使用observesOn选择了另一个。

至于最后一个问题，有可能确定通过查看线程的名字由Schedulers返回通用的标准正确的调度。它们具有可预测的前缀，如RxNewThreadScheduler-1，因此几个String.startsWith()调用可能会隔离正确的调度。但这很脆弱，因为它不能正确处理用户创建的调度程序，并且线程命名方案可能在将来发生变化。

Answer 2

您可以尝试在dismissService()中创建Handler，并且在回调中，您可以回发给关联的活套。

这只有在线程调用到dismissService()的工作原理是弯针线。

编辑：如果你特别想要做一些对IO Scheduler。您也可以从Scheduler创建一名工作人员，并按此方式安排您的操作。