Spring WebFlux 流式数据拉取与推送的实现

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 代表了从下游服务获取的原始数据流。
    Flux<DifficultFaultMessageVo> faults = queryDifficultFaultMessage(paramDto);
    return Flux.create(sink -> {
        // faults.subscribeOn(Schedulers.boundedElastic()): 这行代码至关重要。它告诉上游的 faults 流在 boundedElastic 调度器上执行其订阅操作（即执行网络请求和处理响应）。
        faults.subscribeOn(Schedulers.boundedElastic())
              // 这里创建了一个新的 Flux。create 方法允许我们手动控制如何向这个流中发射数据。我们传入一个 Consumer，它接收一个 FluxSink 对象（这里的参数名为 sink）作为参数。Sink（汇）就是数据流的出口，我们可以通过它发射数据 (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（无限等待，对于流式响应关键！）
        .writeTimeout(120, TimeUnit.SECONDS)   // 写入超时2分钟
        .callTimeout(0, TimeUnit.SECONDS)      // 整个调用超时：0（无限等待）
        .retryOnConnectionFailure(true)        // 自动重试连接失败
        .connectionPool(new ConnectionPool(20, 5, TimeUnit.MINUTES)) // 连接池（20个空闲连接，存活5分钟）
        .build();

private Flux<DifficultFaultMessageVo> queryDifficultFaultMessage(DifficultFaultMessageDto paramDto) {
    return Flux.create(emitter -> { // 这个emitter是内部Flux的Sink
        Request request = buildRequest(paramDto);
        Call call = httpClient.newCall(request);
        
        call.enqueue(new Callback() { // 异步执行HTTP请求
            @Override
            public void onFailure(...) {
                if (!emitter.isCancelled()) {
                    emitter.error(...); // 网络失败，向上游发射错误
                }
            }

            @Override
            public void onResponse(...) throws IOException {
                if (!response.isSuccessful()) {
                    if (!emitter.isCancelled()) {
                        emitter.error(...); // HTTP状态码非2xx，向上游发射错误
                    }
                    return;
                }
                // 成功响应，开始处理流式响应体
                processResponseStream(response, emitter);
            }
        });

        // 重要：注册取消回调
        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());

		// 省略业务代码
        ... 
        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确保资源关闭
        ...
        BufferedSource source = responseBody.source(); // 获取缓冲数据源
        AtomicBoolean isComplete = new AtomicBoolean(false); // 标志位，是否收到结束事件

        try {
            while (!emitter.isCancelled()) { // 循环，只要下游没有取消就继续读
                String line = source.readUtf8Line(); // 读取一行UTF-8文本
                if (line == null) break; // 读到null表示流自然结束（服务器关闭连接）
                if (StringUtils.isBlank(line)) continue; // 忽略空行

                // SSE协议格式：每段数据以"data: "开头
                if (line.startsWith("data:")) {
                    String jsonData = line.substring(6); // 截取"data: "后面的JSON字符串
                    // 过滤：只处理包含"answer"或"message_end"的数据行
                    if (!jsonData.contains("answer") && !jsonData.contains("message_end")) {
                        continue;
                    }

                    DifficultFaultMessageVo vo = processModelResponse(jsonData); // 解析JSON为值对象
                    if (vo != null) {
                        emitter.next(vo); // 解析成功，立即发射给下游（最终到前端）
                        // 如果遇到结束事件，标记完成并结束循环
                        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响应被关闭
    }
}


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

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

            // 3. 创建前端响应对象
            DifficultFaultMessageVo result = new DifficultFaultMessageVo();
            result.setConversationId(rawResponse.getConversationId());

            // 4. 处理结束事件
            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;
        }
    }