如何在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操作符，构建灵活的重试逻辑。

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

代码示例：

Uni-CourseHelper

私人AI助教，高效学习工具

下载

AtomicBoolean hasRetried = new AtomicBoolean(false);

Flux 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标志确保只处理第一个成功响应。

总结

通过retryWhen和onErrorResume，结合标志位控制，我们实现了Spring WebFlux环境下高效的LLM Gateway容灾重试机制，确保服务高可用性，并保障SSE数据流的完整性。 此方案灵活可扩展，适用于各种类型的错误处理和重试策略。