java - 热可观察的 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，因为我不希望 observable 在出现单个异常时停止发出事件。有更好的模式可以使用吗？我想到了 .retry() ，但我不确定这对于热可观察是否有意义？

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

Answer 1

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

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

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

2) 错误可能是可恢复和不可恢复。

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

如果出现 fatal error ，例如您添加监听器失败，因此不可能有进一步的消息，您不应该默默地忽略它。发出 onError 因为你的 observable 无论如何都不起作用。

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

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