在不同的线程rxJava上运行PublishSubject

Question

我正在运行RxJava并创建一个主题以使用onNext（）方法生成数据。我正在使用Spring。

这是我的设置：

@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);
    }
}

在RxJava流上生成新数据的方式是通过Autowire private SubjectObserver SubjectObserver，然后调用SubjectObserver。发布（newDataObjGenerated）

无论我为subscribeOn（）指定了什么

• Schedulers.io（）
• Schedulers.computation（）
• 我的线程
• Schedulers.new线程

onNext（）及其内部的实际工作是在同一个线程上完成的，该线程实际调用主题上的onNext（）来生成/生成数据。

这是正确的吗？如果是，我错过了什么？我希望在不同的线程上完成doSomething（）。

更新

在我的调用类中，如果我更改调用发布方法的方式，那么当然会分配一个新线程供订阅者运行。

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

谢谢，

Answer 1

“Observable”上的每个操作符返回一个具有额外行为的新实例，但是，您的代码只应用subscribeOn和observeOn，然后丢弃它们生成的任何内容并订阅原始主题。您应该链接方法调用，然后订阅：

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();
     }
});

请注意，SubbeOn对Publishsubject没有实际影响，因为它的订阅（）方法中没有发生订阅副作用。