利用Spring WebFlux構(gòu)建LLM網(wǎng)關(guān)的重試機(jī)制

在構(gòu)建LLM網(wǎng)關(guān)時(shí)，需要處理服務(wù)間的通信，并確保當(dāng)某個(gè)服務(wù)不可用時(shí)，能夠無(wú)縫切換到備用服務(wù)。本文將探討如何使用Spring WebFlux實(shí)現(xiàn)這一目標(biāo)，尤其是在網(wǎng)關(guān)到Server B通信失敗時(shí)，如何重試并連接到Server C。

場(chǎng)景描述

我們的LLM網(wǎng)關(guān)調(diào)用鏈路為：客戶端 -> 網(wǎng)關(guān) -> Server B。如果網(wǎng)關(guān)到Server B的連接失敗，我們希望網(wǎng)關(guān)能夠重試并連接到Server C。這需要網(wǎng)關(guān)能夠捕獲到Server B的錯(cuò)誤響應(yīng)碼，并在失敗時(shí)自動(dòng)切換到Server C。

代碼分析及改進(jìn)方案

我們先來(lái)看原始的sseHttp方法，它處理網(wǎng)關(guān)到Server B或Server C的請(qǐng)求：

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 -> {
                    // 錯(cuò)誤處理邏輯
                })
                // ...其他邏輯...</response>

為了實(shí)現(xiàn)重試策略，我們需要捕獲Server B的錯(cuò)誤響應(yīng)碼，并在發(fā)生錯(cuò)誤時(shí)切換到Server C。之前的嘗試存在一些問(wèn)題：簡(jiǎn)單的try-catch無(wú)法捕獲Flux內(nèi)部的錯(cuò)誤；subscribe方法是非阻塞的，導(dǎo)致錯(cuò)誤處理邏輯無(wú)法及時(shí)生效。

最佳實(shí)踐：利用retryWhen和onErrorResume

為了解決上述問(wèn)題，我們應(yīng)該利用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 -> {
                // 記錄錯(cuò)誤日志，方便調(diào)試
                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.")));
}</response>

這段代碼使用onStatus處理HTTP錯(cuò)誤狀態(tài)碼，并使用retryWhen進(jìn)行重試，最多重試3次，每次間隔1秒。 filter確保只重試WebClientResponseException類型的異常。如果重試次數(shù)耗盡，則拋出GatewayException。

然后，在調(diào)用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()); // 記錄錯(cuò)誤日志
            return sseHttp(serverCUrl);
        })
        .next();</response>

這段代碼先嘗試連接Server B，如果發(fā)生WebClientResponseException，則嘗試連接Server C。 next()方法確保只返回一個(gè)結(jié)果。

處理多個(gè)成功響應(yīng)

如果Server B和Server C都成功返回?cái)?shù)據(jù)，我們需要確保只處理一個(gè)響應(yīng)。 可以使用一個(gè)AtomicBoolean變量來(lái)跟蹤是否已經(jīng)成功處理過(guò)響應(yīng)：

AtomicBoolean success = new AtomicBoolean(false);

Flux<response> sseHttp(String url) {
    // ... (previous code) ...
    .doOnNext(response -> {
        if (success.compareAndSet(false, true)) {
            // 處理成功的響應(yīng)
        }
    })
    // ... (rest of the code) ...
}</response>

通過(guò)以上改進(jìn)，我們實(shí)現(xiàn)了更健壯的重試機(jī)制，能夠有效處理服務(wù)間的通信故障，并確保LLM網(wǎng)關(guān)的高可用性。 記住添加充分的日志記錄，方便排查問(wèn)題。

以上是在構(gòu)建LLM gateway時(shí)，如何使用Spring WebFlux實(shí)現(xiàn)從serverB到serverC的重試策略？的詳細(xì)內(nèi)容。更多信息請(qǐng)關(guān)注PHP中文網(wǎng)其他相關(guān)文章！