RxJava io.reactivex.exceptions.OnErrorNotImplementedException：必须从片段主机的主线程调用

Question

我对RxJava中的线程处理有点困惑。

@Override
protected void onCreate(Bundle savedInstanceState) {
    super.onCreate(savedInstanceState);
    setContentView(R.layout.activity_main);

    button = (Button) findViewById(R.id.btn);
    clickObservable = RxView.clicks(button);
    vpPager = findViewById(R.id.vpPager);
    adapterViewPager = new MyPagerAdapter(getSupportFragmentManager());
    vpPager.setAdapter(adapterViewPager);

    //debouncebale button.
    disposable = clickObservable
            .debounce(2000, TimeUnit.MILLISECONDS)
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Consumer<Object>() {
                @Override
                public void accept(Object o) throws Exception {
                    //This prints main thread.
                    Log.d(TAG, " o " + o + " thread " + Thread.currentThread().getName());
                    vpPager.setCurrentItem(vpPager.getCurrentItem() + 1, true);
                }
            });

    pageChangedObservable = RxViewPager.pageSelections(vpPager);
    pageChangedObservable
            .observeOn(AndroidSchedulers.mainThread())
            .delay(2000, TimeUnit.MILLISECONDS)
            .subscribe(new Consumer<Integer>() {
                @Override
                public void accept(final Integer integer) throws Exception {
                    //This prints Thread RxComputationThreadPool-1
                    Log.d(TAG, "Thread " + Thread.currentThread().getName());
                    vpPager.setCurrentItem(integer  + 1, true);
                }
            });



}

第一个Log语句打印主线程，第二个Log语句打印RxComputationThreadPool-1。我不知道为什么。

Answer 1

请注意，一些运营商在指定的默认Schedulers上运行，特别是与时间偏移有关的运营商（例如delay）。否则，如果没有指定Scheduler，他们会在生成下一次发射之前等待主线程。

如果你检查docs，你会看到delay默认运行计算Scheduler。

另外，对于那些有默认Schedulers的运算符，通常会有一个重载变量，允许您指定所需的Scheduler。

Answer 2

实际上，如果您不提供调度程序，则delay（）默认使用Schedulers.computation（）。

在你的第二种情况下，你应该使用

.delay（1，TimeUnit.SECONDS，AndroidSchedulers.mainThread（））

如果你想观察主线程中的结果。或者移动

Opicerven（antraytscetulersmaintaret（））

在.delay（）之后;

当您转到.delay（）方法的定义时，可以找到它。

/ ** *返回一个Observable，它发出源ObservableSource发出的项目，按时间向前移动*指定的延迟。来自源ObservableSource的错误通知不会延迟。 *

Scheduler:

* This version of {@code delay} operates by default on the {@code computation} {@link Scheduler} Hope I answer your question.