RxJava io.reactivex.exceptions.OnErrorNotImplementException:

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 语句打印 RxComputingThreadPool-1。我不知道为什么。

Answer 1

请注意，某些运算符在指定的默认Scheduler上运行，特别是与时间偏移量有关的运算符(例如本例中的delay(。 否则，如果未指定Scheduler，它们将在生成下一个发射之前等待时阻塞主线程。

如果您查看文档，您将看到默认情况下delay对计算Scheduler进行操作。

此外，对于那些默认SchedulerS 的运算符，通常有一个重载变体，允许您指定所需的Scheduler.

Answer 2

实际上，如果您不提供调度程序，Delay(( 默认使用 Schedulers.computing((。

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

.delay(1， TimeUnit.SECONDS， AndroidSchedulers.mainThread(((

如果要在主线程中观察结果。或移动

.observeOn(AndroidSchedulers.mainThread(((

在 .delay(( 之后;

当你转到 .delay(( 方法的定义时，你可以找到它。

/** * 返回一个可观察量，该可观察量发出由源发出的项 可观察源在时间上向前移动 * 指定的延迟。来自源可观察源的错误通知不会延迟。 *

调度程序：

* 此版本的 {@code 延迟} 默认在 {@code 计算} {@link 调度程序} 上运行 希望我回答你的问题。