RSocket遇ConnectionErrorException时如何配置自动重试避免误退出

安全阿岳

2026-8-29

核心结论

你的预期完全合理。当前代码出现keepalive断连直接退出的问题，本质是对RSocket重连机制的作用域理解偏差。

根因说明

你配置的.reconnect(Retry.fixedDelay(...))只作用于连接生命周期的管理层，不会直接接管你手动获取到的单次物理连接的事件：

调用requesterMono.block()拿到的是第一次建连成功的RSocketRequester实例，你后续调用.rsocket().onClose()订阅的是这单个物理连接的关闭事件——不管断连原因是keepalive超时、网络闪断、服务端主动断开、还是致命协议错误，只要这个连接挂了，onClose()就会发出错误信号，完全不会触发你配置的重连逻辑。
RSocket的重连机制逻辑是：当旧连接断开时，自动在内部发起新的建连请求，替换掉失效的旧连接，对外始终暴露一个逻辑上可用的客户端。你手动block拿到单次连接后直接盯它的关闭事件，相当于绕过了重连封装层，直接接在了底层单次连接的事件流上，自然所有断连都会触发退出。
keepalive超时本质是连接失活的常规异常，完全属于可重试范畴，本来就不应该触发客户端退出。

修正方案

不要手动持有单次连接的引用、也不要直接订阅单次连接的onClose()事件，把连接生命周期管理完全交给RSocket的重连层，同时给重连逻辑加错误过滤，区分可重试异常和致命异常：

Mono<RSocketRequester> requesterMono = rsocketRequesterBuilder.setupRoute(SETUP_ROUTE)
    .setupData(new ObjectMapper().writeValueAsString(getAuthInfo()))
    .dataMimeType(MimeTypeUtils.parseMimeType(WellKnownMimeType.APPLICATION_JSON.getString()))
    .rsocketConnector(connector -> connector.acceptor(responder)
        .fragment(MTU_BYTE_SIZE)
        .keepAlive(Duration.ofSeconds(KEEP_ALIVE_INTERVAL_SEC), Duration.ofSeconds(KEEP_ALIVE_MAX_TIME_SEC))
        .reconnect(Retry.fixedDelay(Integer.MAX_VALUE, Duration.ofSeconds(RSOCKET_RETRY_INTERVAL_SECONDS))
            // 过滤异常：仅对可重试异常执行重连
            .filter(throwable -> {
                // 可根据自身业务调整判断逻辑：
                // 认证失败、协议不兼容、配置错误这类重试也无法恢复的异常返回false，终止重连
                // 网络超时、连接重置、keepalive失活这类临时异常返回true，继续重连
                return !(throwable instanceof AuthenticationException) 
                    && !(throwable instanceof RSocketProtocolException);
            })
            .doAfterRetry(retrySignal -> log.warn("RSocket client is reconnecting to get the newest connection...."))
            // 仅当碰到致命异常、重连终止时才触发退出
            .doOnRetryExhausted(signal -> {
                Throwable error = signal.failure();
                String errMsg = "[RSOCKET] connection closed by fatal error, client will exit. msg: " + ExceptionUtils.getRootCauseMessage(error);
                log.error(errMsg, error);
                doExit();
            })
        )
    )
    .connect(TcpClientTransport.create(TcpClient.create()
        .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, sidecarConfig.getRConnectTimeOutMs())
        .option(ChannelOption.SO_KEEPALIVE, true)
        .option(ChannelOption.TCP_NODELAY, true)
        .host(consoleDomain)
        .port(sidecarConfig.getConsoleRSocketPort())
        .secure(ssl -> ssl.sslContext(sslContext))));

// 直接订阅顶层的requesterMono即可，不需要手动block拿单次连接
requesterMono
    .doOnNext(requester -> log.info("[RSOCKET] sidecar client connected to console"))
    .doFinally(signalType -> log.info("[RSOCKET] sidecar client DISCONNECTED to console"))
    .subscribe();

onError与重试机制的关系

你之前订阅到的onError是单次物理连接的错误信号，仅代表当前这一条连接失效，不代表整个客户端运行终结。
RSocket重连层的作用就是拦截这类单次连接的onError信号：如果判定是可重试错误，就吞掉错误、发起新连接，不会把错误抛给上层订阅者；如果判定是致命错误、或者重试完全耗尽，才会把错误向上传播，触发退出逻辑。
你之前的写法直接绕开了重连层，所以所有连接错误都会直接触发退出，重连逻辑根本没有生效的机会。

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