为什么订阅不在新线程中执行？答案

【问题标题】：Why does subscribe not executed in a new thread?为什么订阅不在新线程中执行？
【发布时间】：2015-03-20 14:41:25
【问题描述】：

我有这样的任务：

Observable.just(getMessagesFromDb()).
    subscribeOn(Schedulers.newThread()).
    observeOn(AndroidSchedulers.mainThread()).
    subscribe(incomingMessages -> {
    //do something
    });

其中getMessagesFromDb 是同步获取消息的方法，内部没有多线程。根据 subscribeOn 方法的 RxAndroid 文档：

在指定的调度器上异步订阅这个 Observable 的观察者

还有我的问题——为什么数据库请求在主线程上执行？异步怎么做？

【问题讨论】：

我认为我以前从未见过方法链在末尾带有点。

标签： rx-java rx-android

【解决方案1】：

您在下面看到的是代码的略微修改版本，其中调用了 getMessagesFromDb() 以将返回值分配给变量：

 Object data = getMessagesFromDb();
 Observable.just(data).
    subscribeOn(Schedulers.newThread()).
    observeOn(AndroidSchedulers.mainThread()).
    subscribe(incomingMessages -> {
      //do something
    });

这正是Observable.just 的工作原理，并且getMessagesFromDb() 确实在主线程中被调用（就像你在主线程中调用它一样）。

您必须使用Observable.fromCallable 方法推迟它，如下所示：

Observable.fromCallable(() -> getMessageFromDb()).
           subscribeOn(Schedulers.newThread()).
           observeOn(AndroidSchedulers.mainThread()).
           subscribe(incomingMessages -> {
              //do something
           });

请注意，您可以使用Observable.defer 代替Observable.fromCallable，但它有点复杂：

Observable.defer(() -> Observable.just(getMessageFromDb()))

此响应的先前版本使用Observable.create。但是用这种方法构建一个 Observable 是很复杂的。尽量选择Observable.defer或Observable.fromCallable

【讨论】：

您也可以：Observable.defer(() -> Observable.just(getMessagesFromDb()).subscribeOn(Schedulers.newThread()))
详细而透彻的解释不仅仅对这个例子有所帮助。
这个例子确实很好地解释了defer和just运算符的概念和使用案例。非常感谢。
你将如何在 RxJava2 中做到这一点？调用 defer 或 fromCallable 将不会执行，除非您调用 subscribe 有效地将类型转换为 Disposable。如果我想要一个 Observable/Flowable 怎么办。