用于热可观察的 RxJava 错误处理

Question

我对RxJava很陌生，对模式等有一些疑问。 我正在使用下面的代码创建一个可观察量：

public Observable<Volume> getVolumeObservable(Epic epic) {
return Observable.create(event -> {
try {
listeners.add(streamingAPI.subscribeForChartCandles(epic.getName(), MINUTE, new HandyTableListenerAdapter() {
@Override
public void onUpdate(int i, String s, UpdateInfo updateInfo) {
if (updateInfo.getNewValue(CONS_END).equals(ONE)) {
event.onNext(new Volume(Integer.parseInt(updateInfo.getNewValue(LAST_TRADED_VOLUME))));
}
}
}));
} catch (Exception e) {
LOG.error("Error from volume observable", e);
}
});
}

一切都按预期工作，但我对错误处理有一些问题。 如果我理解正确，这将被视为"热观察"，即无论是否有订阅，事件都会发生(onUpdate 是远程服务器使用的回调，我无法控制)。

我选择不在这里调用 onError，因为我不希望可观察量在出现单个异常时停止发出事件。有没有更好的模式可以使用？.retry() 浮现在脑海中，但我不确定它对热可观察量是否有意义？

此外，在创建订阅时，但在调用第一个 onNext 之前，如何表示可观察量？它只是一个可观察的.empty()

Answer 1

1)你的可观察量不热。区别因素是多个订阅者是否共享同一订阅。Observable.create()为每个订阅者调用订阅函数，即它是冷的。

不过很容易让它变热。只需添加share()运算符即可。它将订阅第一个订阅者，取消订阅最后一个订阅者。不要忘记使用如下内容实现取消订阅功能：

event.setCancellable(() -> listeners.remove(...));

2) 错误可能是可恢复的，也是不可恢复的。

如果您认为错误是可自我恢复的(无需您执行任何操作)，则不应调用onError因为这会杀死您的可观察对象(不会发出进一步的事件)。您可以通过发出附有错误详细信息的特殊Volume消息来通知您的订阅者。

如果错误是致命的，例如您未能添加侦听器，因此可能没有进一步的消息，您不应该默默地忽略它。发出onError因为您的可观察量无论如何都不起作用。

如果错误需要您执行操作(通常是重试或超时重试)，则可以添加retryXxx()运算符之一。在create()之后但在share()之前执行此操作。

3)Observable是具有subscribe()方法的对象。它的确切表示方式取决于您创建它的方法。例如，请参阅create()的源代码。