如何在Spring WebFlux中实现从serverB到serverC的容灾重试机制？-java教程-PHP中文网

如何在Spring WebFlux中实现从serverB到serverC的容灾重试机制？

聖光之護

发布： 2025-03-24 13:16:21

原创

318人浏览过

如何在spring webflux中实现从serverb到serverc的容灾重试机制？

Spring WebFlux构建LLM Gateway的容灾重试方案

本文阐述如何在Spring WebFlux框架下，为LLM Gateway构建高效的容灾重试机制。具体场景：当Gateway到Server B的请求失败时，自动重试Server C，确保客户端(Client A)获得正确响应，即使Server B不可用。方案支持服务器发送事件(SSE)的逐字数据传输。

挑战

Client A通过Gateway访问Server B。若Gateway与Server B连接失败，需要Gateway自动切换至Server C并重试。目标是即使Server B故障，只要Server C可用，Client A也能收到正确结果。此外，需确保SSE数据流的完整性和顺序性。

解决方案：基于retryWhen和onErrorResume的容灾策略

利用Spring WebFlux的retryWhen操作符和onErrorResume操作符，构建灵活的重试逻辑。

错误捕获与重试： retryWhen拦截错误，根据错误类型决定是否重试。若Server C重试仍失败，则将错误信息返回Client A。
避免重复响应： 使用标志位(例如AtomicBoolean)确保仅返回第一次成功的响应，防止Server B和Server C都可用时出现重复响应。

代码示例：

AtomicBoolean hasRetried = new AtomicBoolean(false);

Flux<Response> responseFlux = sseHttp(serverB.getUrl())
    .retryWhen(companion -> companion.flatMap(error -> {
        if (error instanceof GatewayException) {
            // Gateway异常，尝试连接Server C
            return sseHttp(serverC.getUrl())
                .flatMap(serverCResponse -> {
                    hasRetried.set(true);
                    return Flux.just(serverCResponse);
                });
        } else {
            // 其他错误直接返回
            return Flux.error(error);
        }
    }))
    .onErrorResume(error -> {
        // Server C重试失败，返回错误响应给Client A
        return Flux.just(GatewayExceptionHandler.toStreamErrorResponse(
            new GatewayException("Upstream service error.", HttpStatus.INTERNAL_SERVER_ERROR)));
    })
    .doOnNext(response -> {
        if (!hasRetried.get()) {
            // 只处理第一次成功响应
            // ... your original logic here ...
        }
    });

登录后复制

此示例中，retryWhen捕获Server B的错误，并尝试连接Server C。hasRetried标志确保只处理第一个成功响应。