我对 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,因为我不希望 observable 在出现单个异常时停止发出事件。有更好的模式可以使用吗?我想到了 .retry() ,但我不确定这对于热可观察是否有意义?
此外,在创建订阅时、调用第一个 onNext 之前,可观察对象是如何表示的?它只是一个 Observable.empty()
最佳答案
1) 您的可观察值不热。区别因素是多个订阅者是否共享相同的订阅。 Observable.create()
为每个订阅者调用订阅函数,即它是冷。
不过让它变热很容易。只需添加 share()
运算符即可。它将向第一个订阅者订阅并与最后一个订阅者取消订阅。不要忘记使用如下方式实现取消订阅功能:
event.setCancellable(() -> listeners.remove(...));
2) 错误可能是可恢复和不可恢复。
如果您认为错误是可自行恢复的(您无需采取任何操作),则不应调用onError
,因为这会杀死您的可观察对象(不会发出进一步的事件)。您可以通过发出附加错误详细信息的特殊 Volume
消息来通知您的订阅者。
如果出现 fatal error ,例如您添加监听器失败,因此不可能有进一步的消息,您不应该默默地忽略它。发出 onError
因为你的 observable 无论如何都不起作用。
如果出现错误需要您执行操作(通常是重试或超时重试),您可以添加 retryXxx()
运算符之一。在create()
之后但在share()
之前执行此操作。
3) Observable
是一个具有 subscribe()
方法的对象。它的具体表示方式取决于您创建它的方法。例如,请参见 create()
的源代码。
关于java - 热可观察的 RxJava 错误处理,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/43339635/