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創建一名工作人員，並按此方式安排您的操作。