利用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方法，加入重试逻辑：
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网关的高可用性。 记住添加充分的日志记录，方便排查问题。
