如何检测RSocket连接或重连是否成功以实现健康检查？

核心原理说明 #

你当前用到的doBeforeRetry/doAfterRetry属于Reactor Retry操作符的钩子，只能捕获重连流程的启动/等待阶段事件，拿不到真正的连接成功信号——因为Retry操作符只负责在流异常时触发重连动作，连接建立完成的生命周期信号，是由底层RSocketConnector统一对外暴露的，不属于Retry链路的事件范畴。

具体实现改造 #

在你现有代码的rsocketConnector配置块中，新增框架内置的连接生命周期钩子即可，不需要依赖Retry链路的回调判断，示例代码如下：

// 提前定义全局健康状态标记，供健康检查程序读取
private final AtomicBoolean rsocketConnectionHealth = new AtomicBoolean(false);

requester = RSocketRequester.builder()
    .rsocketConnector(connector -> {
        // 保留原有重连配置
        connector.reconnect(Retry
            .fixedDelay(Integer.MAX_VALUE, Duration.ofSeconds(1))
            .doBeforeRetry(e -> System.out.println("doBeforeRetry===>" + e))
            .doAfterRetry(e -> System.out.println("doAfterRetry===>" + e))
        );
        connector.payloadDecoder(PayloadDecoder.ZERO_COPY);

        // 新增：连接/重连成功的回调，首次建连、重连成功都会触发
        connector.onConnected(rsocket -> {
            System.out.println("RSocket连接/重连成功，连接实例ID：" + rsocket.hashCode());
            // 更新健康状态为正常
            rsocketConnectionHealth.set(true);
        });

        // 新增：连接断开的回调，和健康检查逻辑配套
        connector.onClose(rsocket -> {
            System.out.println("RSocket连接断开");
            // 更新健康状态为异常
            rsocketConnectionHealth.set(false);
        });
    })
    .dataMimeType(MediaType.APPLICATION_CBOR)
    .rsocketStrategies(strategies)
    .tcp("localhost", 7999);

关键注意事项 #

• onConnected的触发时机是RSocket协议握手全部完成的节点，和你实际可发起请求的状态完全一致，不会出现TCP连通但RSocket协商失败的“假健康”问题
• 不需要在业务请求链路里额外做连接状态判断，健康检查程序直接读取rsocketConnectionHealth的原子值即可，实现轻量无额外请求开销
• 不要尝试在Retry链路上用doOnNext/doOnSuccess捕获连接成功事件，重连返回的是持续监听的流，不是单次连接完成的信号，很容易出现回调不触发的问题

内容的提问来源于stack exchange，提问作者Saji