利用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<Response> 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方法，加入重试逻辑：
AI 舌诊面诊问诊医疗系统 + 诊疗商城系统源码是一套结合 AI 辅助诊断与商品转化的健康类程序，基于浩创云、ThinkPHP（PHP）开发，支持微信服务号与 H5 端，适合做中医舌诊面诊问答、健康建议推送以及配套诊疗商品的运营。
Flux<Response> 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<Response> 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<Response> sseHttp(String url) {
// ... (previous code) ...
.doOnNext(response -> {
if (success.compareAndSet(false, true)) {
// 处理成功的响应
}
})
// ... (rest of the code) ...
}
通过以上改进，我们实现了更健壮的重试机制，能够有效处理服务间的通信故障，并确保LLM网关的高可用性。 记住添加充分的日志记录，方便排查问题。
