Spring WebFlux实现ServerB到ServerC重试策略详解

利用Spring WebFlux构建LLM网关的重试机制

在构建LLM网关时，需要处理服务间的通信，并确保当某个服务不可用时，能够无缝切换到备用服务。本文将探讨如何使用Spring WebFlux实现这一目标，尤其是在网关到Server B通信失败时，如何重试并连接到Server C。

场景描述

我们的LLM网关调用链路为：客户端 -> 网关 -> Server B。如果网关到Server B的连接失败，我们希望网关能够重试并连接到Server C。这需要网关能够捕获到Server B的错误响应码，并在失败时自动切换到Server C。

代码分析及改进方案

我们先来看原始的sseHttp方法，它处理网关到Server B或Server C的请求：

Flux responseFlux = webClient.create(url)
                .post()
                .headers(httpHeaders -> setHeaders(httpHeaders, headers))
                .contentType(MediaType.APPLICATION_JSON)
                .bodyValue(jsonBody)
                .retrieve()
                .onStatus(status -> status != HttpStatus.OK, response -> {
                    // 错误处理逻辑
                })
                // ...其他逻辑...

为了实现重试策略，我们需要捕获Server B的错误响应码，并在发生错误时切换到Server C。之前的尝试存在一些问题：简单的try-catch无法捕获Flux内部的错误；subscribe方法是非阻塞的，导致错误处理逻辑无法及时生效。

最佳实践：利用retryWhen和onErrorResume

为了解决上述问题，我们应该利用Spring WebFlux提供的retryWhen和onErrorResume操作符。

首先，修改sseHttp方法，加入重试逻辑：

Flux sseHttp(String url) {
    return webClient.create(url)
            .post()
            .headers(httpHeaders -> setHeaders(httpHeaders, headers))
            .contentType(MediaType.APPLICATION_JSON)
            .bodyValue(jsonBody)
            .retrieve()
            .onStatus(HttpStatus::isError, clientResponse -> {
                // 记录错误日志，方便调试
                return Mono.error(new WebClientResponseException("Server returned error status: " + clientResponse.rawStatusCode(), clientResponse.rawStatusCode(), clientResponse.headers().asHttpHeaders(), clientResponse.bodyToMono(String.class).block(), null));
            })
            .bodyToFlux(typeRef)
            .retryWhen(Retry.backoff(3, Duration.ofSeconds(1))
                    .filter(throwable -> throwable instanceof WebClientResponseException)
                    .onRetryExhaustedThrow((spec, signal) -> new GatewayException("Failed to connect to both Server B and Server C after multiple retries.")));
}

这段代码使用onStatus处理HTTP错误状态码，并使用retryWhen进行重试，最多重试3次，每次间隔1秒。 filter确保只重试WebClientResponseException类型的异常。如果重试次数耗尽，则抛出GatewayException。

然后，在调用sseHttp的地方，使用onErrorResume处理Server B的失败，并切换到Server C：

Mono responseMono = sseHttp(serverBUrl)
        .onErrorResume(WebClientResponseException.class, ex -> {
            log.warn("Failed to connect to Server B: {}", ex.getMessage()); // 记录错误日志
            return sseHttp(serverCUrl);
        })
        .next();

这段代码先尝试连接Server B，如果发生WebClientResponseException，则尝试连接Server C。 next()方法确保只返回一个结果。

处理多个成功响应

如果Server B和Server C都成功返回数据，我们需要确保只处理一个响应。 可以使用一个AtomicBoolean变量来跟踪是否已经成功处理过响应：

AtomicBoolean success = new AtomicBoolean(false);

Flux sseHttp(String url) {
    // ... (previous code) ...
    .doOnNext(response -> {
        if (success.compareAndSet(false, true)) {
            // 处理成功的响应
        }
    })
    // ... (rest of the code) ...
}

通过以上改进，我们实现了更健壮的重试机制，能够有效处理服务间的通信故障，并确保LLM网关的高可用性。 记住添加充分的日志记录，方便排查问题。