SubscibeOnが私の理解あたりとして被験者

Question

には影響しません、

• SubscribeOnオペレータは 観察可能ソースは、発光を開始するべきスレッドを指定します。
• チェーンに複数のSubscribeOnがある場合、最初のフローはフロー全体に対して の効果を得ます。
• ObserveOnが鎖中にたびObserveOn存在する、任意 時点で下流にスレッドを反転するために使用することができ、それは下流

ためのスレッドを変更しかし、私はSubjectとIとサンプルを試しチェーン全体でSubscribeOnの効果が見えません。

は、ここで私は

   mSubject 
      .map(s -> "String :" + s) 
      .doOnNext(s -> Log.d(TAG, "Started at Thread :" + Thread.currentThread().getName())) 
      .flatMap(s -> Observable.just(1)) 
      .map(Object::toString) 
      .subscribeOn(Schedulers.io()) 
      .observeOn(AndroidSchedulers.mainThread()) 
      .subscribe(s -> { 
       Log.d(TAG, "subscribed: " + s + " at " + Thread.currentThread().getName()); 

      }); 

と

ここ

  mSubject.onNext("hello"); 

あなたが見ることができますが、私は.subscribeOn(Schedulers.io())を与えているいくつかの他の場所からのトリガ、だから私のようにそれを消費

Subject<String> mSubject = PublishSubject.create(); 

私のサンプルです期待値

   mSubject 
      .map(s -> "String :" + s) 
      .doOnNext(s -> Log.d(TAG, "Started at Thread :" + Thread.currentThread().getName())) 
      .flatMap(s -> Observable.just(1)) 
      .map(Object::toString) 

これらがScheduler Threadで実行されるまで。それから私はobserveOnを使ってスレッドをメインスレッドにフリップさせます。しかし、ここにこのコードのログ

   D/MainActivity: Started at Thread :main 
       D/MainActivity: subscribed: 1 at main 

なぜスケジューラスレッドで起動しませんでしたか？

件名を使用せずに普通のObservableを使ってみました。

   Observable.just("Hello") 
      .map(s -> "String :" + s) 
      .doOnNext(s -> Log.d(TAG, "Started at Thread :" + Thread.currentThread().getName())) 
      .flatMap(s -> Observable.just(1)) 
      .map(Object::toString) 
      .subscribeOn(Schedulers.io()) 
      .observeOn(AndroidSchedulers.mainThread()) 
      .subscribe(s -> { 
       Log.d(TAG, "subscribed: " + s + " at " + Thread.currentThread().getName()); 

      }); 

、ここでは、ログ

   D/MainActivity: Started at Thread :RxCachedThreadScheduler-2 
      D/MainActivity: subscribed: 1 at main 

私は期待通りに働いている見ることができるのです！

（公式RxJava Subjectドキュメントによってリンクされている）、ポストUsing subjectsから被験者

Answer 1

で何が起こったので、それは述べて：デフォルトで

を、被験者は、スレッド間で任意の同期を実行しないでください。彼らはスケジューラーを取るのではなく、すべてのシリアライゼーションと文法上の正確さが主語の呼び出し側によって処理されると仮定します。

これは、使用されたスレッドがonNext()を呼び出すコードの1つであり、後で観測スレッドに送信されることを意味します。