rx java - Android RxJava, Non Blocking?

Question

我的理解是rxjava-android在一个单独的线程上执行操作(当提供正确的Scheduler时)，导致非阻塞操作，但是一个快速而肮脏的测试似乎证明这是不正确的。

我使用了以下代码片段，在这两种情况下，UI都被阻塞了…

片段1

Observable observable = Observable.create(new Observable.OnSubscribe<Object>() {
  @Override
  public void call(Subscriber<? super Object> subscriber) {
    int i = 0;
    while (i == 0) {}
    subscriber.onCompleted();
  }
});
observable.subscribeOn(Schedulers.newThread());
observable.observeOn(AndroidSchedulers.mainThread());
observable.subscribe();

片段2

Observable observable = Observable.create(new Observable.OnSubscribe<Object>() {
  @Override
  public void call(Subscriber<? super Object> subscriber) {
    SystemClock.sleep(5000);
    subscriber.onCompleted();
  }
});
observable.subscribeOn(Schedulers.newThread());
observable.observeOn(AndroidSchedulers.mainThread());
observable.subscribe();

我错过了什么吗?

Answer 1

错误在于您使用了错误的Observable。正确的代码应该是:

Observable observable = Observable.create(new Observable.OnSubscribe<Object>() {
  @Override
  public void call(Subscriber<? super Object> subscriber) {
    int i = 0;
    while (i == 0) {}
    subscriber.onCompleted();
  }
});
observable.subscribeOn(Schedulers.newThread()).observeOn(AndroidSchedulers.mainThread()).subscribe();

subscribeOn和observeOn都返回一个新的Observable来实现它们的功能。但是原始的Observable没有被修改。实际上，每个操作符都会创建一个新的Observable，而不需要修改原来的Observable。