java - 如何检测重试 Flux 中的恢复？

Question

我已经成功设置了一个 Flux 从远程系统接收事件(协议(protocol)是 websocket，但这与问题无关)并使用 retryBackoff 方法优雅地处理连接故障。代码(简化)是这样的:

Flux flux = myEventFlux
            .retryBackoff( Long.MAX_VALUE, Duration.ofSeconds(5) )
            .publish()
            .refCount();

flux.subscribe( System.out::println );

现在我想处理连接丢失和连接恢复事件，以便在 UI 中显示一些提示，或者至少注册一些日志。检测错误似乎很容易，只需在 retryBackoff 之前执行 doOnError 即可。但恢复是另一回事......我需要类似“错误后第一个成功事件”的东西，所以我尝试了这个:

flux.next().subscribe( event -> System.out.println("first = " + event) );

它在第一次正常连接(没有先前错误)时有效，但在错误后的后续重新连接中无效。

Answer 1

您想要实现的困难在于，从以下角度来看，无法区分订阅重试 Flux 的两个合法行尾订阅者与触发两次尝试的一个订阅者myEventFlux。

因此，您可以使用doOnSubscribe(或自3.2.10.RELEASE起的doFirst)，但它会受到上述限制。它还会在原始连接上触发，而不仅仅是重试...

也许对于 UI 用例这仍然会有帮助？