渴望订阅– RxJava常见问题解答

在教学和指导RxJava以及撰写本书之后，我注意到某些领域尤其成问题。我决定发布一些简短的提示，以解决最常见的陷阱。这是第一部分。
Observable和Flowable本质上是惰性的。这意味着无论您在Flowable放置了多么繁琐或长时间运行的逻辑，仅当有人订阅时，它才会被评估。并且还有某人订阅的次数。下面的代码段对此进行了说明：

private static String slow() throws InterruptedException {logger.info("Running");TimeUnit.SECONDS.sleep(1);return "abc";
}//...Flowable flo = Flowable.fromCallable(this::slow);
logger.info("Created");
flo.subscribe();
flo.subscribe();
logger.info("Done");

这样的Observable或Flowable将不可避免地打印：

19:37:57.368 [main] - Created
19:37:57.379 [main] - Running
19:37:58.383 [main] - Running
19:37:59.388 [main] - Done

请注意，您需要两次支付sleep()的价格sleep()两次订阅）。此外，所有逻辑都在客户端（ main ）线程中运行，除非通过subscriptionOn subscribeOn()请求或异步流隐式可用，否则RxJava中没有隐式线程。问题是：我们是否可以热切地强制运行订阅逻辑，以便每当有人订阅该流时，就已经对其进行了预先计算或至少开始了计算？

完全渴望评估

最明显但有缺陷的解决方案是急于计算流返回的任何内容，并简单地将其包装为固定的Flowable ：

Flowable<String> eager() {final String slow = slow();return Flowable.just(slow);
}

不幸的是，这大大破坏了RxJava的目的。首先，像subscribeOn()这样的运算符将不再起作用，并且无法将计算卸载到其他线程。更糟糕的是，即使eager()返回了Flowable但按照定义，它将始终阻止客户端线程。很难推理，组合和管理此类流。通常，即使需要进行急切的评估，也应避免使用这种模式，而应选择延迟加载。

使用

下一个示例仅使用cache()运算符：

Flowable<String> eager3() throws InterruptedException {final Flowable<String> cached =Flowable.fromCallable(this::slow).cache();cached.subscribe();return cached;
}

这个想法很简单：用惰性Flowable包装计算并缓存它。 cache()运算符所做的是，它会记住第一次订阅时发出的所有事件，以便在出现第二个Subscriber ，它将接收相同的事件缓存序列。但是cache()运算符（像大多数其他运算符一样）是惰性的，因此我们必须第一次强制订阅。调用subscribe()将预填充缓存，此外，如果第二个订户出现在slow()计算完成之前，它将同样等待它，而不是第二次启动它。

此解决方案有效，但请记住，由于未涉及Scheduler ， subscribe()实际上将被阻止。如果要在后台预填充Flowable ，请尝试subscribeOn() ：

Flowable<String> eager3() throws InterruptedException {final Flowable<String> cached =Flowable.fromCallable(this::slow).subscribeOn(justDontAlwaysUse_Schedulers.io()).cache();cached.subscribe();return cached;
}

是的，在生产系统上使用Schedulers.io()存在问题且难以维护，因此请避免使用自定义线程池。

错误处理

令人遗憾的是，吞噬RxJava中的异常非常容易。如果slow()方法失败，这就是我们上一个示例中可能发生的情况。异常不会完全被吞没，但是默认情况下，如果没有人对此感兴趣，它将在System.err上打印堆栈跟踪。同样，未处理的异常也包装在OnErrorNotImplementedException 。如果您执行任何形式的集中式日志记录，则不太方便，很可能会丢失。您可以使用doOnError()操作进行日志记录，但它仍然通过例外下游RxJava认为未处理的为好，一次包装与OnErrorNotImplementedException 。因此，让我们在subscribe()实现onError回调：

Flowable<String> eager3() throws InterruptedException {final Flowable<String> cached =Flowable.fromCallable(this::slow).cache();cached.subscribe(x -> {/* ignore */},e -> logger.error("Prepopulation error", e));return cached;
}

我们不想处理实际事件，而只是处理subscribe()错误。此时，您可以安全地返回此类Flowable 。急切且有希望的是，只要您订阅了它，数据就已经可用。注意，例如，Hystrix的observe()方法也很急切，而懒惰的toObservable()相反。这是你的选择。