ObserveOn不工作

Question

我开始为Android学习Rx，并且出现一个错误。这是我的代码：

在我Subscriber我打印记录当前线程的名称：

Subscriber<Integer> integerSubscriber = new Subscriber<Integer>() { 
     // ... onCompleted, onError 
     @Override 
     public void onNext(Integer s) { 
      Log.e("RX", "threadName " + getCurrThreadName()); 
     } 
};  

在这里，我想在后台运行的代码，并在mainThread得到观察数据：

ArrayList list = new ArrayList(); 
list.add(...) // creating data list 

Observable.from(list) 
    .map(TransformFunc.getTransformer()) 
    .subscribeOn(AndroidSchedulers.mainThread()) 
    .observeOn(Schedulers.from(executor)) 
    .subscribe(integerSubscriber); 

... 

其他代码： static String getCurrThreadName（）{ return Thread.currentThread（）。getName（）; }

static class TransformFunc implements Func1<String, Integer> { 

    private static TransformFunc instance; 

    static TransformFunc getTransformer() { 
     ... // return instance 

     @Override 
     public Integer call(@NonNull String s) { 
      ... 
       TimeUnit.SECONDS.sleep(4); 
       Log.e("RX", "threadName " + getCurrThreadName()); 
      ... 

      return s.length(); 
     } 
} 

但是当我在设备上运行此代码，我看到白色屏幕5-10秒，这在日志：

12-02 16:34:39.374 26086-26086/com.shevart.fitnessnotes E/RX: threadName main 
12-02 16:34:39.378 26086-26180/com.shevart.fitnessnotes E/RX: threadName pool-1-thread-2 
12-02 16:34:43.375 26086-26086/com.shevart.fitnessnotes E/RX: threadName main 
12-02 16:34:43.379 26086-26216/com.shevart.fitnessnotes E/RX: threadName pool-1-thread-3 
12-02 16:34:47.376 26086-26086/com.shevart.fitnessnotes E/RX: threadName main 
12-02 16:34:47.396 26086-26253/com.shevart.fitnessnotes E/RX: threadName pool-1-thread-4 
12-02 16:34:51.376 26086-26086/com.shevart.fitnessnotes E/RX: threadName main 
12-02 16:34:51.398 26086-26292/com.shevart.fitnessnotes E/RX: threadName pool-1-thread-5 

为什么地图call()之前调用onNext()？

Answer 1

你想观察主线程，而不是周围的其他方式。

.subscribeOn(Schedulers.from(executor)) 
.observeOn(AndroidSchedulers.mainThread()) 

Answer 2

通过调用.subscribeOn()你基本上告知Observable做在一个特定的Scheduler工作。另一方面，您通过.observeOn()告诉Observable向您发送该线程中的所有通知。

所以你需要交换你的订阅和观察调度。

.subscribeOn(Schedulers.from(executor)) 
.observeOn(AndroidSchedulers.mainThread())