SpringBoot中使用Flux实现流式返回的方法小结

本文主要是介绍SpringBoot中使用Flux实现流式返回的方法小结，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

《SpringBoot中使用Flux实现流式返回的方法小结》文章介绍流式返回（StreamingResponse）在SpringBoot中通过Flux实现,优势包括提升用户体验、降低内存消耗、支持长连...

背景

近期在使用deepseek/openai等网页和APP时，发现大模型在思考和回复时，内容是一点点的显示出来的，于是好奇他们的实现方式。经调研和使用开发者工具抓取请求，每次聊天会向后台发送一个http请求，而这个接口跟普通接口一次性返回不一样，而是以流式的返回。

流式返回的核心概念与优势

在传统的 Web 开发中，接口通常以「一次性返回完整响应体」的形式工作。而 ** 流式返回（Streaming Response）** 指的是服务器在处理请求时，将响应结果分段逐步返回给客户端，而非等待所有数据生成完成后再一次性返回。这种模式具有以下核心优势：

1. 提升用户体验

对于大数据量响应（如文件下载、长文本流）或实时交互场景（如聊天机器人对话），客户端可边接收数据边处理，减少「空白等待时间」，提升实时性感知。

2. 降低内存消耗

服务器无需在内存中缓存完整响应数据，尤其适合处理高并发、大流量场景，降低 OOM（内存溢出）风险。

3. 支持长连接与实时通信

天然适配实时数据推送场景（如日志监控、股票行情更新），可与 SSE（Server-Sent Events）、WebSocket 等技术结合使用。

大模型的接口，尤其是那些带推理的模型接口返回，数据就是一点点的返回的，因此如果要提升用户体验，最好的方式就是采用流式接口返回。

在SpringBoot中基于Flux的流式接口实现

1. 依赖配置

在 pom.XML 中引入 WebFlux 依赖：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-webflux</artifactId>
</dependency>

2. 流式接口实现（以模拟大模型对话为例）

import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Flux;
@RestController
@RequestMapping("/api/chat")
public class ChatController {
    @PostMapping(produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<String> streamChat(@RequestBody ChatRequest request) {
        // 调用大模型 API 并返回 Flux 流
        return callLargeModelApi(request.mandroidessage())
            .doOnNext(chunk -> log.info("发送响应片段: {}", chunk))
            .doOnError(error -> log.error("流式处理出错", error));
    }
    // 模拟调用大模型 API，返回 Flux 流
    private Flux<String> callLargeModelApi(String prompt) {
        // 实际项目中需替换为真实的大模型调用逻辑
        return Flux.just(
            "您好！", 
            "我是您的AI助手。", 
            "您的问题是：" + prompt, 
            "我将为您提供详细解答..."
        )
        .delayElements(Duration.ofMillis(300)); // 模拟实时响应延迟
    }
}

3. 关键配置说明

响应格式：设置 produces = MediaType.TEXT_EVENT_STREAM_VALUE，符合 SSE 协议。
异步处理：Flux 流中的元素会被自动转换为 SSE 格式（data: <China编程;内容>\n\n）并推送至客户端。
背压控制：通过 onBackpressureBuffer() 或 onBackpressureDrop() 处理客户端消费速率问题。

浏览器端 JS 调用方案

1. 使用 EventSource（简化版）

function connectWithEventSource() {
    const source = new EventSource("/api/chat");
    const chatWindow = document.getElementById("chat-window");
    source.onmessage = (event) => {
        chatWindow.innerhtml += `<div>${event.data}</div>`;
        chatWindow.scrollTop = chatWindow.scrollHeight;
    };
    source.onerror = (error) => {
        console.error("EventSource failed:", error);
        source.close();
    };
}

2. 使用 Fetch API（支持 POST 请求）

async function connectWithFetch() {
    const response = await fetch("/api/chat", {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify({ message: "你好" })
    });
    const reader = response.body.getReader();
    const decoder = new TextDecoder();
    const chatWindow = document.getElementById("chat-window");
    while (true) {
        const { done, value } = await reader.read();
        if (done) break;
        // 解码并处理数据块
        const chunk = decoder.decode(value, { stream: true });
        const messages = chunk.split('\n\n')
            .filter(line => line.trim().startsWith('data:'))
            .map(line => line.replace('data:', '').trim());
        messages.forEach(msg => {
            chatWindow.innerHTML += `<div>${msg}</div>`;
            chatWindow.scrollTop = chatWindow.scrollHeight;
        });
    }
}

调用Deepseek模型实战

写一个接口，通过Spring AI Alibaba ，调用阿里云百炼的deepseek模型，返回Flux流数据

基础使用详见：快速开始-阿里云SWiyJbpring AI Alibaba官网官网

这里只给出转Flux的示例，即通过client/model的stream方法来转，并通过map方法将每个流转成前端需要的数据(我这里是区分了thinking思考和content的数据，便于前端显示)：

    public Flux<ChatMessageResponse> processRealMessage(ChatMessageRequest request) throws ChatBaseException {
        // 获取会话的历史消息
        List<Message> messages = new ArrayList<>();
        List<ChatMessage> chatMessages = this.chatMessageService.getConversationMessage(request.getSessionId(), 1, 20);
        for (ChatMessage chatMessage : chatMessages) {
            if (Constants.MESSAGE_ROLE_USER.equals(chatMessage.getRole())) {
                messages.add(new UserMessage(chatMessage.getContent()));
            } else {
                messages.add(new AssistantMessage(chatMessage.getContent()));
            }
        }
        // js记录用户的输入
        ChatMessage message = new ChatMessage();
        message.setContent(request.getContent());
        message.setType("text");
        message.setRole(Constants.MESSAGE_ROLE_USER);
        chatMessageService.insertMessage(request.getSessionId(), message);
        StringBuilder sb = new StringBuilder();
        // 模拟流式响应
        return this.chatClient.prompt().messages(messages).user(request.getContent()).stream().chatResponse().doOnNext(response -> {
                    String content = response.getResult().getOutput().getText();
                    if (StringUtils.isNotBlank(contChina编程ent)) {
                        // 记录完整的响应对象
                        sb.append(content);
                    }
                })
                // 在流结束时记录完整的会话内容
                .doOnComplete(() -> {
                    // 这里记录消息到数据库
                    String content = sb.toString();
                    LOGGER.info("收到模型原始响应结束: " + content);
                    ChatMessage assistantMessage = new ChatMessage();
                    assistantMessage.setContent(content);
                    assistantMessage.setType("text");
                    assistantMessage.setRole(Constants.MESSAGE_ROLE_ASSISTENT);
                    try {
                        chatMessageService.insertMessage(request.getSessionId(), assistantMessage);
                    } catch (ChatBaseException e) {
                        LOGGER.error("processMessage2 doOnComplete insertMessage error");
                    }
                }).map(response -> {
                    String content = response.getResult().getOutput().getText();
                    String thinking = response.getResults().get(0).getOutput().getMetadata().get("reasoningContent").toString();
                    if (StringUtils.isNotEmpty(content)) {
                        LOGGER.info("content" + content);
                        return new ChatMessageResponse("content", content);
                    } else if (StringUtils.isNotEmpty(thinking)) {
                        LOGGER.info("thinking" + thinking);
                        return new ChatMessageResponse("thinking", thinking);
                    } else {
                        LOGGER.info("done~~~~");
                        return new ChatMessageResponse("done", "");
                    }
                });
    }

完整代码：MaDiXin/madichat

到此这篇关于SpringBoot中使用Flux实现流式返回的技术总结的文章就介绍到这了,更多相关SpringBoot Flux流式返回内容请搜索China编程(www.chinasem.cn)以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程China编程(www.chinasem.cn)！

这篇关于SpringBoot中使用Flux实现流式返回的方法小结的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！