TL;DR: 将 observeOn(AndroidSchedulers.mainThread())
放到 filter(...)
之后。
subscribeOn(...)
用于指定 Observable
开始操作的线程。对 subscribeOn
的后续调用将被忽略。
因此,如果你写了以下代码,所有内容 将在 Schedulers.newThread()
上执行:
myService.fetchSomeIntegersFromServer()
.subscribeOn(Schedulers.newThread())
.filter(integer -> {
System.out.println(Looper.getMainLooper().getThread() == Thread.currentThread());
return true;
})
.subscribe(integer1 -> { doSomething(integer1); });
当然,你并不想这样:你想在主线程上进行doSomething
操作。
这就是observeOn
的作用所在。在observeOn
后面的所有操作都会在该调度器上执行。因此,在你的例子中,filter
是在主线程上执行的。
相反,将observeOn
移动到subscribe
之前:
myService.fetchSomeIntegersFromServer()
.subscribeOn(Schedulers.newThread())
.filter(integer -> {
System.out.println(Looper.getMainLooper().getThread() == Thread.currentThread());
return true;
})
.observeOn(AndroidSchedulers.mainThread())
.subscribe(integer1 -> { doSomething(integer1) });
现在,filter
将在“新线程”上执行,而doSomething
将在主线程上执行。
为了进一步操作,您可以多次使用 observeOn
:
myService.fetchSomeIntegersFromServer()
.subscribeOn(Schedulers.newThread())
.observeOn(Schedulers.computation())
.filter(integer -> {
System.out.println(Looper.getMainLooper().getThread() == Thread.currentThread());
return true;
})
.observeOn(AndroidSchedulers.mainThread())
.subscribe(integer1 -> { doSomething(integer1) });
在这种情况下,获取将在一个新线程上进行,过滤将在一个计算线程上进行,而doSomething
将在主线程上进行。
请查看ReactiveX - SubscribeOn operator获取官方文档。
observeOn
(以及其他方法)不像构建器模式一样返回相同的Observable
实例,而是返回一个“包装”“旧”的Observable”的新
Observable`实例。 - nhaarman.observeOn(AndroidSchedulers.mainThread())
行。(目前我正在缓存可观察对象,并且我认为第一个observeOn()
调用会处理。) - WonderCsabo