查询 RxJava 线程调度答案

【问题标题】：Query on RxJava thread scheduling查询 RxJava 线程调度
【发布时间】：2023-04-05 01:21:02
【问题描述】：

我是 RxJava2 的新手。在下面的代码中，我无法理解订阅者是如何在后台线程上工作的，即使 Observable/Flowable 在主线程上发出并且没有指定调度程序（使用 subscribeOn(Schedulers.*) 调用）。完整代码见this github repo.

@OnClick(R.id.btn_start_simple_polling)
public void onStartSimplePollingClicked() {

    _log("onStartSimplePollingClicked called on ");  //MAIN THREAD

    final int pollCount = POLL_COUNT;

    Disposable d = Observable
          .interval(INITIAL_DELAY, POLLING_INTERVAL, TimeUnit.MILLISECONDS)
          .map(this::_doNetworkCallAndGetStringResult)
          .take(pollCount)
          .doOnSubscribe(subscription -> {
              _log(String.format("Start simple polling - %s", _counter));     //MAIN THREAD
          })
          .subscribe(taskName -> {
              _log(String.format(Locale.US,
                                 "Executing polled task [%s] now time : [xx:%02d]",
                                 taskName,
                                 _getSecondHand()));
          });

    _disposables.add(d);
}

private String _doNetworkCallAndGetStringResult(long attempt) {
    try {
        _log("_doNetworkCallAndGetStringResult called on "); //BACKGROUND THREAD
        if (attempt == 4) {
            // randomly make one event super long so we test that the repeat logic waits
            // and accounts for this.
            Thread.sleep(9000);
        }
        else {
            Thread.sleep(3000);
        }

    } catch (InterruptedException e) {
        Timber.d("Operation was interrupted");
    }
    _counter++;

    return String.valueOf(_counter);
}

【问题讨论】：

标签： android rx-java rx-java2

【解决方案1】：

由于您没有指定订阅 RxJava 的调度程序，因此默认为同步订阅。所以对onSubscribe 和doOnSubscribe 的调用发生在主线程上。

但是，Observable.interval 运算符需要隐式或显式调度程序来广播onNext 事件。由于您没有指定调度程序，因此默认为Schedulers.computation()。间隔触发后，它继续在同一计算线程上调用_doNetworkCallAndGetStringResult，因此发生在后台。

【讨论】：

【解决方案2】：

默认情况下，RxJava 同步运行，但 @Kiskae 等一些运算符已经告诉您间隔、延迟或其他一些运算符

如果您想异步运行管道，则必须使用observerOn，一旦将管道放入管道中，它将在另一个线程中运行管道

 /**
     * Once that you set in your pipeline the observerOn all the next steps of your pipeline will be executed in another thread.
     * Shall print
     * First step main
     * Second step RxNewThreadScheduler-2
     * Third step RxNewThreadScheduler-1
     */
    @Test
    public void testObservableObserverOn() throws InterruptedException {
        Subscription subscription = Observable.just(1)
                .doOnNext(number -> System.out.println("First step " + Thread.currentThread()
                        .getName()))
                .observeOn(Schedulers.newThread())
                .doOnNext(number -> System.out.println("Second step " + Thread.currentThread()
                        .getName()))
                .observeOn(Schedulers.newThread())
                .doOnNext(number -> System.out.println("Third step " + Thread.currentThread()
                        .getName()))
                .subscribe();
        new TestSubscriber((Observer) subscription)
                .awaitTerminalEvent(100, TimeUnit.MILLISECONDS);
    }

或使用 subscribeOn 它将使您的管道在您指定的线程中运行

/**
 * Does not matter at what point in your pipeline you set your subscribeOn, once that is set in the pipeline,
 * all steps will be executed in another thread.
 * Shall print
 * First step RxNewThreadScheduler-1
 * Second step RxNewThreadScheduler-1
 */
@Test
public void testObservableSubscribeOn() throws InterruptedException {
    Subscription subscription = Observable.just(1)
            .doOnNext(number -> System.out.println("First step " + Thread.currentThread()
                    .getName()))
            .subscribeOn(Schedulers.newThread())
            .doOnNext(number -> System.out.println("Second step " + Thread.currentThread()
                    .getName()))
            .subscribe();
    new TestSubscriber((Observer) subscription)
            .awaitTerminalEvent(100, TimeUnit.MILLISECONDS);
}

你可以在这里看到更多关于异步 rxJava 的例子https://github.com/politrons/reactive/blob/master/src/test/java/rx/observables/scheduler/ObservableAsynchronous.java

【讨论】：