利用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ù)。本文將探討如何使用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)方案

我們先來看原始的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。之前的嘗試存在一些問題：簡(jiǎn)單的try-catch無法捕獲Flux內(nèi)部的錯(cuò)誤； subscribe方法是非阻塞的，導(dǎo)致錯(cuò)誤處理邏輯無法及時(shí)生效。

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

為了解決上述問題，我們應(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變量來跟蹤是否已經(jīng)成功處理過響應(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>

通過以上改進(jìn)，我們實(shí)現(xiàn)了更健壯的重試機(jī)制，能夠有效處理服務(wù)間的通信故障，並確保LLM網(wǎng)關(guā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)文章！