異なるスレッドでPublishSubjectを実行するrxJava

Question

私はRxJavaを実行し、データを生成するためにonNext()メソッドを使用するサブジェクトを作成しています。私は春を使用しています。

これは私の設定です：新しいデータがRxJavaストリーム上で生成され

@Component 
public class SubjectObserver { 
    private SerializedSubject<SomeObj, SomeObj> safeSource; 
    public SubjectObserver() { 
     safeSource = PublishSubject.<SomeObj>create().toSerialized(); 
     **safeSource.subscribeOn(<my taskthreadExecutor>);** 
     **safeSource.observeOn(<my taskthreadExecutor>);** 
     safeSource.subscribe(new Subscriber<AsyncRemoteRequest>() { 
      @Override 
      public void onNext(AsyncRemoteRequest asyncRemoteRequest) { 
      LOGGER.debug("{} invoked.", Thread.currentThread().getName()); 
      doSomething(); 
      } 
     } 
    } 
    public void publish(SomeObj myObj) { 
     safeSource.onNext(myObj); 
    } 
} 

方法は@Autowire private SubjectObserver subjectObserver であり、その後、どんなに私がsubscribeOn() & observeOn()に指定したものをsubjectObserver.publish(newDataObjGenerated)

を呼び出しません：

• Schedulers.io（）
• Schedulers.computation（）
• 私のスレッド
• Schedulers.newThread

onNext()とその中に実際の作業は、実際には/農産物を生成するために、被写体にonNext()を呼び出すのと同じスレッド上で実行されますデータ。

これは間違いありませんか？もしそうなら、私は何が欠けていますか？私はdoSomething()が別のスレッドで実行されることを期待していました。私の呼び出し元のクラスで

更新は

、私は新しいスレッドが上で実行するために、加入者に割り当てられている、その後の進路、publishメソッドを呼び出しています方法を変更した場合。

taskExecutor.execute(() -> subjectObserver.publish(newlyGeneratedObj)); 

おかげで、

Answer 1

余分な行動で新しいインスタンスを返すObservable/Subject上の各オペレータは、しかし、あなたのコードだけでsubscribeOnとobserveOnを適用し、その後、彼らは生産と生Subjectに加入しているものは何でも捨て。

safeSource = PublishSubject.<AsyncRemoteRequest>create().toSerialized(); 

safeSource 
.subscribeOn(<my taskthreadExecutor>) 
.observeOn(<my taskthreadExecutor>) 
.subscribe(new Subscriber<AsyncRemoteRequest>() { 
    @Override 
    public void onNext(AsyncRemoteRequest asyncRemoteRequest) { 
     LOGGER.debug("{} invoked.", Thread.currentThread().getName()); 
     doSomething(); 
    } 
}); 

注そのsubscribe()方法で起こって何のサブスクリプションの副作用がないためsubscribeOnがPublishSubjectには実用的な効果を持っていないこと：あなたは、チェーン、メソッドの呼び出しとサブスクライブする必要があります。