rx 订阅与主题不起作用？

Question

代码如下:

Observable<Integer> observable = Observable.just(1, 2);
observable.doOnNext(i -> System.out.println("Emitting " + i + " on thread " + Thread.currentThread().getName()))
          .subscribeOn(Schedulers.computation())
          .doOnNext(i -> System.out.println("Emitting " + i + " after subscribeOn" + " on thread " + Thread.currentThread().getName()))
          .observeOn(AndroidSchedulers.mainThread())
          .subscribe(i -> System.out.println( "Receiving " + i + " on thread " + Thread.currentThread().getName()));

它给了我这样的输出:

I/System.out: Emitting 1 on thread RxComputationScheduler-1
I/System.out: Emitting 1 after subscribeOn on thread RxComputationScheduler-1
I/System.out: Emitting 2 on thread RxComputationScheduler-1
I/System.out: Emitting 2 after subscribeOn on thread RxComputationScheduler-1
I/System.out: Receiving 1 on thread main
I/System.out: Receiving 2 on thread main

一切正常。

现在我想对Subject做同样的事情，我已经改变了代码如下:

PublishSubject<Integer> subject = PublishSubject.create();
subject.doOnNext(i -> System.out.println("Emitting " + i + " on thread " + Thread.currentThread().getName()))
       .subscribeOn(Schedulers.computation())
       .doOnNext(i -> System.out.println("Emitting " + i + " after subscribeOn" + " on thread " + Thread.currentThread().getName()))
       .observeOn(AndroidSchedulers.mainThread())
       .subscribe(i -> System.out.println( "Receiving " + i + " on thread " + Thread.currentThread().getName()));
subject.onNext(1);

，在这种情况下我没有得到输出。为什么如此?以及如何在给定的调度程序中强制主题发出流?

Answer 1

决定是:

PublishSubject<Integer> subject = PublishSubject.create();
subject
      **.observeOn(Schedulers.computation())**
      .doOnNext(i -> System.out.println("Emitting " + i + " on thread " + Thread.currentThread().getName()))                    
      .observeOn(AndroidSchedulers.mainThread())
      .subscribe(i -> System.out.println( "Receiving " + i + " on thread " + Thread.currentThread().getName()));
subject.onNext(1);

或者你可以从Schedulers.computation()线程