data:{"conversationId":"eef6023a-6ace-420f-ad46-324aa53decf8","event":"agent_message","answer":""}

data:{"conversationId":"eef6023a-6ace-420f-ad46-324aa53decf8","event":"agent_message","answer":""}

data:{"conversationId":"eef6023a-6ace-420f-ad46-324aa53decf8","event":"agent_message","answer":"故障"}

data:{"conversationId":"eef6023a-6ace-420f-ad46-324aa53decf8","event":"agent_message","answer":"根"}

data:{"conversationId":"eef6023a-6ace-420f-ad46-324aa53decf8","event":"agent_message","answer":"因"}
....
data:{"conversationId":"eef6023a-6ace-420f-ad46-324aa53decf8","event":"agent_message","answer":"：\n\n"}

data:{"conversationId":"eef6023a-6ace-420f-ad46-324aa53decf8","event":"message_end","answer":"一"}```

public Flux<DifficultFaultMessageVo> streamingQueryDifficultFault(DifficultFaultMessageDto paramDto) {
    // 這個 Flux 代表了從下游服務(wù)獲取的原始數(shù)據(jù)流。
    Flux<DifficultFaultMessageVo> faults = queryDifficultFaultMessage(paramDto);
    return Flux.create(sink -> {
        // faults.subscribeOn(Schedulers.boundedElastic()): 這行代碼至關(guān)重要。它告訴上游的 faults 流在 boundedElastic 調(diào)度器上執(zhí)行其訂閱操作（即執(zhí)行網(wǎng)絡(luò)請求和處理響應(yīng)）。
        faults.subscribeOn(Schedulers.boundedElastic())
              // 這里創(chuàng)建了一個新的 Flux。create 方法允許我們手動控制如何向這個流中發(fā)射數(shù)據(jù)。我們傳入一個 Consumer，它接收一個 FluxSink 對象（這里的參數(shù)名為 sink）作為參數(shù)。Sink（匯）就是數(shù)據(jù)流的出口，我們可以通過它發(fā)射數(shù)據(jù) (next)、錯誤 (error) 或完成信號 (complete)。
                .subscribe(sink::next, sink::error, sink::complete);
    });
}

private final OkHttpClient httpClient = new OkHttpClient.Builder()
        .connectTimeout(120, TimeUnit.SECONDS) // 連接超時2分鐘
        .readTimeout(0, TimeUnit.SECONDS)      // 讀取超時：0（無限等待，對于流式響應(yīng)關(guān)鍵?。?
        .writeTimeout(120, TimeUnit.SECONDS)   // 寫入超時2分鐘
        .callTimeout(0, TimeUnit.SECONDS)      // 整個調(diào)用超時：0（無限等待）
        .retryOnConnectionFailure(true)        // 自動重試連接失敗
        .connectionPool(new ConnectionPool(20, 5, TimeUnit.MINUTES)) // 連接池（20個空閑連接，存活5分鐘）
        .build();

private Flux<DifficultFaultMessageVo> queryDifficultFaultMessage(DifficultFaultMessageDto paramDto) {
    return Flux.create(emitter -> { // 這個emitter是內(nèi)部Flux的Sink
        Request request = buildRequest(paramDto);
        Call call = httpClient.newCall(request);
        
        call.enqueue(new Callback() { // 異步執(zhí)行HTTP請求
            @Override
            public void onFailure(...) {
                if (!emitter.isCancelled()) {
                    emitter.error(...); // 網(wǎng)絡(luò)失敗，向上游發(fā)射錯誤
                }
            }

            @Override
            public void onResponse(...) throws IOException {
                if (!response.isSuccessful()) {
                    if (!emitter.isCancelled()) {
                        emitter.error(...); // HTTP狀態(tài)碼非2xx，向上游發(fā)射錯誤
                    }
                    return;
                }
                // 成功響應(yīng)，開始處理流式響應(yīng)體
                processResponseStream(response, emitter);
            }
        });

        // 重要：注冊取消回調(diào)
        emitter.onCancel(() -> {
            if (!call.isCanceled()) {
                call.cancel(); // 如果下游取消訂閱（如客戶端斷開），則取消OkHttp請求
            }
        });
    });
}

    private Request buildRequest(DifficultFaultMessageDto paramDto) {
        // 2.組裝請求頭信息
        MediaType parse = MediaType.parse("application/json;charset=UTF-8");
        // 3.組裝請求體信息
        JSONObject requestBody = new JSONObject();
        requestBody.put("x x x", paramDto.getxxxType());
        requestBody.put("apiKey", paramDto.getApiKey());

		// 省略業(yè)務(wù)代碼
        ... 
        log.info("API請求體: {}", requestBody);
        return new Request.Builder().url("http://xxx.x.xx.xx:8000/servicexxx/r/postApi").post(body)
                .addHeader("X-APP-ID", "xx09xx3xx0d7").addHeader("X-APP-KEY", "9jfksjfjkxxkkssdc")
                .addHeader("Content-Type", "application/json").build();
    }

private void processResponseStream(Response response, FluxSink<DifficultFaultsVo> emitter) {
    try (ResponseBody responseBody = response.body()) { // 使用try-with-resources確保資源關(guān)閉
        ...
        BufferedSource source = responseBody.source(); // 獲取緩沖數(shù)據(jù)源
        AtomicBoolean isComplete = new AtomicBoolean(false); // 標(biāo)志位，是否收到結(jié)束事件

        try {
            while (!emitter.isCancelled()) { // 循環(huán)，只要下游沒有取消就繼續(xù)讀
                String line = source.readUtf8Line(); // 讀取一行UTF-8文本
                if (line == null) break; // 讀到null表示流自然結(jié)束（服務(wù)器關(guān)閉連接）
                if (StringUtils.isBlank(line)) continue; // 忽略空行

                // SSE協(xié)議格式：每段數(shù)據(jù)以"data: "開頭
                if (line.startsWith("data:")) {
                    String jsonData = line.substring(6); // 截取"data: "后面的JSON字符串
                    // 過濾：只處理包含"answer"或"message_end"的數(shù)據(jù)行
                    if (!jsonData.contains("answer") && !jsonData.contains("message_end")) {
                        continue;
                    }

                    DifficultFaultMessageVo vo = processModelResponse(jsonData); // 解析JSON為值對象
                    if (vo != null) {
                        emitter.next(vo); // 解析成功，立即發(fā)射給下游（最終到前端）
                        // 如果遇到結(jié)束事件，標(biāo)記完成并結(jié)束循環(huán)
                        if ("message_end".equals(vo.getEvent())) {
                            isComplete.set(true);
                            emitter.complete();
                            break;
                        }
                    }
                }
            }
        } catch (Exception e) {
            // 處理讀取和解析過程中的異常
            if (!emitter.isCancelled()){
                emitter.error(e);
            }
        } finally {
            // 確保流最終完成
            if (!isComplete.get() && !emitter.isCancelled()){
                emitter.complete();
            }
        }
        ...
    } catch (Exception e) {
        ...
    } finally {
        response.close(); // 最終確保HTTP響應(yīng)被關(guān)閉
    }
}


private DifficultFaultMessageVo processModelResponse(String jsonData) {
        try {
            // 1. 解析JSON
            DifficultFaultMessageVo rawResponse = JSONUtil.toBean(
                    jsonData,
                    DifficultFaultMessageVo.class,
                    false
            );

            // 2. 過濾非消息事件
            String event = rawResponse.getEvent();

            // 3. 創(chuàng)建前端響應(yīng)對象
            DifficultFaultMessageVo result = new DifficultFaultMessageVo();
            result.setConversationId(rawResponse.getConversationId());

            // 4. 處理結(jié)束事件
            if ("message_end".equals(event)) {
                result.setEvent("message_end");
                result.setAnswer("");
                return result;
            }
            result.setAnswer(rawResponse.getAnswer());
            result.setEvent(rawResponse.getEvent());

            return result;

        } catch (Exception e) {
            log.error("JSON解析錯誤: {} - {}", jsonData, e.getMessage());
            return null;
        }
    }