实战指南：Spring AI Alibaba + QWen 流式输出在Java中的高效应用
1. 为什么你需要关注Spring AI Alibaba和流式输出？
如果你是一个Java开发者，最近肯定被各种大模型（LLM）的消息刷屏了。看着别人用Python三两行代码就能调用AI模型，生成酷炫的内容，是不是心里有点痒，又觉得Java生态里好像缺了点什么？别急，这种感觉我懂。以前想在Java项目里集成一个AI能力，要么得自己吭哧吭哧去封装HTTP客户端，处理复杂的JSON，要么就得依赖一些不那么“Spring”风格的第三方库，用起来总感觉别扭。
这就是Spring AI Alibaba出现的原因。它就像是Spring生态给Java开发者发的一个“AI大礼包”。简单来说，它把调用大模型（比如阿里的通义千问QWen）这件事，变得跟我们在Spring里用JdbcTemplate操作数据库一样自然。你不用再关心底层的网络请求、认证、序列化这些脏活累活，只需要关注你的业务逻辑：给模型一个提示（Prompt），然后处理它返回的结果。
而流式输出（Streaming），则是这个“大礼包”里最让人兴奋的“黑科技”。想象一下这个场景：你问模型“给我写一篇关于春天的散文”。如果没有流式输出，你的前端页面会一直转圈圈，直到模型把整篇几百字的散文全部生成完毕，才一次性吐给你。用户等待的这好几秒钟里，可能已经失去耐心关掉页面了。
但有了流式输出，情况就完全不同了。模型会像真人打字一样，一个字、一个词、一句话地实时“流”回给你的应用。前端页面可以立刻开始渲染这些内容，用户几乎在提问后的一瞬间就能看到“春回大地，万物复苏...”这样的开头，体验变得无比流畅。这种“边生成边返回”的能力，对于构建聊天机器人、代码补全、内容创作助手这类需要即时反馈的应用来说，是提升用户体验的关键。接下来，我就带你一步步拆解，如何用最“Spring”的方式，在Java里轻松玩转这套组合拳。
2. 手把手搭建你的第一个流式AI应用
理论说再多，不如动手跑一遍。我保证，跟着下面的步骤，10分钟内你就能拥有一个能和你“打字聊天”的Spring Boot应用。我们先从最基础的环境准备开始。
2.1 项目初始化与核心依赖
首先，确保你的“地基”是稳固的。我推荐使用 start.spring.io 来快速生成项目，记得选上 Spring Boot 3.3.x 或更高版本，以及 JDK 17 或以上。这两个是硬性要求，Spring AI Alibaba的某些特性依赖这些新版本提供的支持。
创建好项目后，打开你的pom.xml文件。这里有个关键点：因为Spring AI Alibaba的里程碑版本（M2）还没进中央仓库，我们需要手动添加Spring的仓库地址。别担心，这就像告诉Maven：“去这几个特殊的仓库找找看”。
<repositories>
<repository>
<id>sonatype-snapshots</id>
<url>https://oss.sonatype.org/content/repositories/snapshots</url>
<snapshots>
<enabled>true</enabled>
</snapshots>
</repository>
<repository>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/milestone</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</repository>
</repositories>
添加完仓库，就可以引入最重要的“主角”了——spring-ai-alibaba-starter。同时，为了实现流式响应，我们需要spring-boot-starter-webflux，它提供了响应式编程的支持，是Flux流的基础。当然，传统的Web starter也一并加上，方便我们测试。
<dependencies>
<!-- Spring AI Alibaba 核心依赖 -->
<dependency>
<groupId>com.alibaba.cloud.ai</groupId>
<artifactId>spring-ai-alibaba-starter</artifactId>
<version>1.0.0-M2</version>
</dependency>
<!-- 响应式Web，支持Flux流式返回 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-webflux</artifactId>
</dependency>
<!-- 传统Web，用于基础HTTP服务 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
</dependencies>
2.2 获取通行证：阿里云API Key配置
依赖搞定，接下来我们需要一把“钥匙”来访问通义千问模型。这把钥匙就是阿里云百炼平台的API Key。你只需要有一个阿里云账号（没有的话注册一个很快），然后按下面步骤操作：
- 登录阿里云官网，进入 “百炼” 产品页面。
- 在控制台找到 “模型服务” 或 “API密钥管理”。
- 开通“百炼大模型推理”服务（新用户通常有免费额度，比如通义千问就有100万Token，足够你玩很久了）。
- 创建一个新的API Key，并把它复制保存好，就像保存密码一样。
拿到Key后，怎么告诉我们的应用呢？最安全方便的做法是把它设置成环境变量。在Mac或Linux的终端里，可以这样设置：
export AI_DASHSCOPE_API_KEY=你的真实API Key
在Windows的PowerShell里则是：
$env:AI_DASHSCOPE_API_KEY="你的真实API Key"
当然，你也可以直接在application.properties或application.yml里写死，但强烈不推荐，尤其是项目要上传到GitHub等公共仓库时，这会泄露你的密钥。
# application.properties - 不推荐直接写死，建议用环境变量
spring.ai.dashscope.api-key=${AI_DASHSCOPE_API_KEY}
2.3 编写流式对话控制器
环境配好了，钥匙也有了，现在来写最核心的业务代码。我们要创建一个Controller，它接收用户的一句话，然后以数据流的形式返回模型的回复。
import org.springframework.ai.chat.ChatClient;
import org.springframework.ai.chat.prompt.Prompt;
import org.springframework.ai.chat.prompt.PromptTemplate;
import org.springframework.core.io.Resource;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Flux;
import java.util.Map;
@RestController
@RequestMapping("/ai")
@CrossOrigin(origins = "*") // 允许前端跨域调用
public class StreamChatController {
private final ChatClient chatClient;
// 注入一个Prompt模板文件（可选，后面会讲）
@Value("classpath:prompts/chat.st")
private Resource promptResource;
// 通过构造器注入ChatClient
public StreamChatController(ChatClient.Builder builder) {
this.chatClient = builder.build();
}
@GetMapping(value = "/chat/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<String> streamChat(@RequestParam String message) {
// 1. 构建Prompt（这里先用简单方式，后面讲模板）
String userPrompt = "请用友好、简洁的语气回答以下问题：" + message;
Prompt prompt = new Prompt(userPrompt);
// 2. 调用ChatClient，获取流式响应
return chatClient.prompt(prompt)
.stream() // 关键！这开启了流式模式
.content(); // 提取每一块响应中的文本内容
}
}
我来解释一下这段代码的精华：
@GetMapping注解里的produces = MediaType.TEXT_EVENT_STREAM_VALUE是灵魂。它告诉Spring，这个接口返回的是 Server-Sent Events (SSE) 格式的数据流，这是浏览器原生支持的一种流式协议。ChatClient是Spring AI Alibaba提供的统一客户端，你不需要知道背后是调用的通义千问还是其他模型，它帮你屏蔽了差异。.stream()方法调用是关键转折点。如果不调用它，默认是call()方法，会阻塞等待全部结果。一调用.stream()，返回的就变成了一个Flux<ChatResponse>，也就是一个响应式数据流。.content()是一个便捷方法，它从每一块ChatResponse中提取出文本内容，最终我们得到一个Flux<String>，即字符串流。
启动你的Spring Boot应用，然后用浏览器或者curl命令测试一下：
curl -N http://localhost:8080/ai/chat/stream?message=介绍一下Java
你会看到文字不是一个完整的JSON一次性返回，而是一行一行、几乎实时地显示在终端里。这就是流式输出的魔力！
3. 从“能用”到“好用”：Prompt工程与高级流式处理
基础功能跑通了，但你可能发现回复有点“机械”。怎么让AI的回答更符合你的业务场景？怎么在流式返回中处理更复杂的信息？这部分我们来深入一下。
3.1 使用Prompt模板：让AI更懂你
直接拼接字符串构建Prompt，在简单场景下没问题，但复杂起来就很难维护。Spring AI提供了强大的PromptTemplate功能，它支持在模板文件中定义结构，然后动态填充变量。
首先，在src/main/resources目录下创建一个prompts文件夹，然后新建一个文件 chat.st (.st是Spring Template的缩写)：
你是一个专业的Java技术专家，回答问题时需要满足以下要求：
1. 语言风格：{style}。
2. 如果问题涉及代码，请用{language}语言举例。
3. 如果问题比较宽泛，请先给出核心要点，再展开说明。
用户的问题是：{input}
请开始你的回答：
看，这个模板里定义了三个变量：style, language, input。接下来，我们改造一下Controller，使用这个模板：
@GetMapping(value = "/chat/stream/adv", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<String> streamChatWithTemplate(@RequestParam String input,
@RequestParam(defaultValue = "严谨专业") String style,
@RequestParam(defaultValue = "Java") String language) {
// 1. 加载模板文件
PromptTemplate promptTemplate = new PromptTemplate(promptResource);
// 2. 创建参数Map，填充模板变量
Map<String, Object> model = Map.of(
"input", input,
"style", style,
"language", language
);
// 3. 根据模板和参数，创建最终的Prompt对象
Prompt prompt = promptTemplate.create(model);
// 4. 流式调用
return chatClient.prompt(prompt)
.stream()
.content();
}
现在，你可以这样调用接口，获得一个用Python举例、风格幽默的答案：
http://localhost:8080/ai/chat/stream/adv?input=解释一下多态&style=幽默风趣&language=Python
通过模板，你可以轻松管理不同场景下的Prompt，比如客服话术、代码审查规则、文案生成风格等，让AI的输出质量大幅提升。
3.2 驾驭Flux流：控制、转换与错误处理
Flux是Project Reactor中的核心响应式类型，代表一个0到N个元素的异步序列。把它单纯当作一个返回类型有点“浪费”了。我们可以像操作Java 8的Stream一样，对Flux进行各种中间操作。
场景一：我想在流式输出中，给每一句话前面加个前缀。
public Flux<String> streamChatWithPrefix(@RequestParam String message) {
Prompt prompt = new Prompt(message);
return chatClient.prompt(prompt)
.stream()
.map(chatResponse -> "AI说: " + chatResponse.getResult().getOutput().getContent());
// 使用map操作符转换每个元素
}
场景二：我觉得模型返回得太快了，想模拟“人”的思考速度，每段之间加个短暂延迟。
public Flux<String> streamChatWithDelay(@RequestParam String message) {
Prompt prompt = new Prompt(message);
return chatClient.prompt(prompt)
.stream()
.delayElements(Duration.ofMillis(100)) // 每个元素延迟100毫秒
.content();
}
场景三：流式调用可能失败（网络超时、API限额等），必须要有健壮的错误处理。
public Flux<String> streamChatSafe(@RequestParam String message) {
Prompt prompt = new Prompt(message);
return chatClient.prompt(prompt)
.stream()
.onErrorResume(throwable -> {
// 记录错误日志
log.error("调用AI模型失败", throwable);
// 返回一个友好的错误信息流，而不是让整个请求崩溃
return Flux.just("抱歉，AI助手暂时开小差了，请稍后再试。");
})
.content();
}
场景四：我不想无限制地流下去，最多只接收10条响应片段。
public Flux<String> streamChatWithLimit(@RequestParam String message) {
Prompt prompt = new Prompt(message);
return chatClient.prompt(prompt)
.stream()
.take(10) // 只取前10个元素
.content();
}
这些操作符的组合使用，让你能精细地控制整个流式响应的行为，适应各种复杂的业务需求。
4. 前端如何“接住”这个流？完整全栈实战
后端流式接口准备好了，如果前端还是用普通的Ajax请求，那等于白忙活。前端也需要用能处理流的技术来“接住”数据。这里我给出两种最主流的方案。
4.1 方案一：使用EventSource (SSE) - 最简单
Server-Sent Events (SSE) 是HTML5的标准，浏览器原生支持，使用起来非常简单。它特别适合服务器向客户端单向推送数据的场景，比如我们的AI回复流。
前端JavaScript代码示例：
<!DOCTYPE html>
<html>
<body>
<input type="text" id="question" placeholder="输入你的问题...">
<button onclick="askAI()">提问</button>
<div id="answer" style="white-space: pre-wrap; border:1px solid #ccc; padding:10px; min-height:100px;"></div>
<script>
function askAI() {
const question = document.getElementById('question').value;
const answerDiv = document.getElementById('answer');
answerDiv.innerHTML = ''; // 清空旧答案
// 构建带参数的URL
const url = `http://localhost:8080/ai/chat/stream?message=${encodeURIComponent(question)}`;
// 创建EventSource对象连接流式端点
const eventSource = new EventSource(url);
// 监听'message'事件，这是接收数据的主要事件
eventSource.onmessage = function(event) {
// event.data 就是服务器发来的每一块文本
answerDiv.innerHTML += event.data;
// 自动滚动到底部，方便阅读
answerDiv.scrollTop = answerDiv.scrollHeight;
};
// 监听'error'事件，处理错误或连接关闭
eventSource.onerror = function(err) {
console.error('EventSource failed:', err);
eventSource.close(); // 关闭连接
answerDiv.innerHTML += '\n\n--- 对话结束或发生错误 ---';
};
// 可以监听连接打开事件（可选）
eventSource.onopen = function() {
console.log('连接已建立，开始接收流式数据...');
};
}
</script>
</body>
</html>
SSE的优点就是简单，几乎零配置。但缺点是它只支持服务器到客户端的单向通信，并且有最大连接数限制（浏览器通常每个域名限制6个）。对于大多数AI问答场景，这完全够用了。
4.2 方案二：使用Fetch API + ReadableStream - 更灵活
如果你需要更底层的控制，或者后端不是严格的SSE格式（比如是纯文本流），那么使用Fetch API配合ReadableStream是更强大的选择。
前端JavaScript代码示例：
<script>
async function askAIWithFetch() {
const question = document.getElementById('question').value;
const answerDiv = document.getElementById('answer');
answerDiv.innerHTML = 'AI正在思考...';
const url = `http://localhost:8080/ai/chat/stream?message=${encodeURIComponent(question)}`;
try {
const response = await fetch(url);
const reader = response.body.getReader();
const decoder = new TextDecoder('utf-8');
answerDiv.innerHTML = ''; // 清空“正在思考”提示
while (true) {
const { done, value } = await reader.read();
if (done) {
console.log('流式响应结束');
break;
}
// 解码并追加数据块
const chunk = decoder.decode(value, { stream: true });
answerDiv.innerHTML += chunk;
answerDiv.scrollTop = answerDiv.scrollHeight;
}
} catch (error) {
console.error('请求失败:', error);
answerDiv.innerHTML = '请求失败，请检查网络或控制台。';
}
}
</script>
这种方式的控制力更强，你可以处理任何格式的流数据。但代码相对复杂一些，需要手动处理读取器和解码。
4.3 结合Vue/React现代框架
在实际的Vue或React项目中，你通常会在组件中封装这个流式请求逻辑。这里以React函数组件为例，展示一个更工程化的写法：
import React, { useState } from 'react';
function AIChatBox() {
const [input, setInput] = useState('');
const [answer, setAnswer] = useState('');
const [isLoading, setIsLoading] = useState(false);
const [eventSource, setEventSource] = useState(null);
const handleAsk = () => {
if (eventSource) {
eventSource.close(); // 关闭之前的连接
}
setAnswer('');
setIsLoading(true);
const url = `http://your-backend/ai/chat/stream?message=${encodeURIComponent(input)}`;
const es = new EventSource(url);
setEventSource(es);
es.onmessage = (event) => {
setIsLoading(false);
// 使用函数式更新，确保拿到最新的answer状态
setAnswer(prev => prev + event.data);
};
es.onerror = () => {
setIsLoading(false);
es.close();
setAnswer(prev => prev + '\n\n--- 连接已关闭 ---');
};
};
const handleStop = () => {
if (eventSource) {
eventSource.close();
setEventSource(null);
setIsLoading(false);
}
};
return (
<div>
<input value={input} onChange={(e) => setInput(e.target.value)} />
<button onClick={handleAsk} disabled={isLoading}>提问</button>
<button onClick={handleStop}>停止</button>
<div style={{ whiteSpace: 'pre-wrap' }}>{isLoading ? '思考中...' : answer}</div>
</div>
);
}
这个组件增加了“停止”按钮，可以主动中断流式请求，用户体验更好。在实际项目中，你还需要考虑错误处理的UI展示、请求防抖、历史记录管理等。
5. 避坑指南与性能优化实战
东西做出来了，但要稳定、高效地用在生产环境，还有几个坑你得提前知道。这些都是我趟过雷、踩过坑总结出来的经验。
5.1 常见问题与排查
问题一：连接超时或流意外中断。 流式连接是长连接，比普通HTTP请求更脆弱。网络波动、代理服务器、负载均衡器都可能提前关闭连接。
- 后端解决：调整Web容器的超时设置。对于Spring Boot内嵌的Tomcat或Netty，可以在
application.properties中增加：# 增加连接保持活动状态的时间（秒） server.connection-timeout=60s # 对于WebFlux（Netty），可以设置响应超时 spring.webflux.client.response-timeout=60s - 前端解决：实现自动重连机制。在EventSource的
onerror回调中，判断错误类型，如果不是致命错误，可以等待几秒后重新建立连接，并从断点处继续请求（这需要后端支持记录上下文）。
问题二：流式响应内容乱码或格式不对。 确保前后端的编码一致，都是UTF-8。在Controller上明确指定produces = MediaType.TEXT_EVENT_STREAM_VALUE，Spring会帮你处理好SSE的格式（每一条消息以data: 开头，以两个换行符\n\n结尾）。如果使用Fetch API，也要确保正确使用TextDecoder。
问题三：高并发下性能问题。 每个流式连接都会占用一个线程（或反应式线程池中的资源）。虽然WebFlux是异步非阻塞的，但大量并发长连接仍然对服务器资源是考验。
- 优化方向：合理设置线程池。对于WebFlux，默认的线程数可能偏少，可以根据服务器核心数调整：
# 调整反应式Netty的工作线程数 server.netty.connection-pool.max-connections=1000 server.netty.connection-pool.acquire-timeout=45s - 根本方案：对于超大规模并发，考虑引入消息队列（如Kafka, RabbitMQ） 或 专门的消息推送服务。架构可以变为：用户请求 -> 后端快速返回一个任务ID -> 后端异步调用AI模型并将结果分片写入消息队列 -> 前端通过另一个长连接或WebSocket订阅该任务ID的消息流。这样能将计算密集的AI调用与高并发的连接管理解耦。
5.2 监控与可观测性
线上系统，不能是黑盒。你需要知道流式接口的健康状况。
-
日志记录：在Controller方法的关键位置（开始、结束、异常）添加日志。但注意，不要记录整个流的内容（可能很长且包含敏感信息），可以记录请求的元数据，如用户ID、问题长度、模型名称、流式耗时等。
@GetMapping("/chat/stream") public Flux<String> streamChat(@RequestParam String message, @RequestHeader("User-Id") String userId) { log.info("收到流式聊天请求，用户: {}, 问题长度: {}", userId, message.length()); long startTime = System.currentTimeMillis(); return chatClient.prompt(new Prompt(message)) .stream() .doOnNext(chunk -> log.debug("发送数据块，大小: {}", chunk.length())) .doOnError(error -> log.error("流式处理失败，用户: {}", userId, error)) .doOnComplete(() -> { long duration = System.currentTimeMillis() - startTime; log.info("流式请求完成，用户: {}, 总耗时: {}ms", userId, duration); }) .content(); }这里用了
doOnNext,doOnError,doOnComplete等操作符，它们像“切面”一样，让你在不改变主数据流的情况下，插入监控逻辑。 -
指标监控：集成Micrometer和Prometheus，暴露关键指标。
ai.request.count: 总请求数ai.request.duration: 请求耗时分布ai.stream.chunk.count: 每个流返回的数据块数量ai.stream.active.connections: 当前活跃的流式连接数 这些指标能帮你快速定位是模型服务变慢，还是你的应用服务器压力过大。
5.3 成本控制与限流
通义千问的API不是免费的（虽然有慷慨的免费额度）。如果你的应用面向公众，必须考虑成本控制和防止滥用。
-
用户级限流：使用Spring的
@RateLimiter注解（需集成Resilience4j等库）或Guava的RateLimiter，对每个用户/IP在一段时间内的请求次数进行限制。// 伪代码示例：使用Bucket4j进行令牌桶限流 @GetMapping("/chat/stream") public Flux<String> streamChat(@RequestParam String message, HttpServletRequest request) { String ip = request.getRemoteAddr(); if (!rateLimiter.tryConsume(ip, 1)) { // 每秒最多1个令牌 return Flux.error(new TooManyRequestsException("请求过于频繁，请稍后再试")); } // ... 正常处理逻辑 } -
Token计数与预算：更精细的控制是估算每次请求消耗的Token数（输入+输出）。你可以在Prompt里要求模型在回复结束时附上本次消耗的Token数（部分API支持），或者在流式返回结束后，根据返回的文本长度进行粗略估算。为每个用户设置每日/每月Token预算，并在接近限额时进行提醒或拒绝服务。
-
缓存策略：对于常见、重复的问题（例如“你好”、“你是谁”），没必要每次都调用昂贵的模型。可以在调用模型前，先查一下Redis等缓存。如果命中，甚至可以将缓存的完整答案模拟成流式（按词或按句分块）返回，用户体验完全一致，但成本为零。
流式输出不仅仅是技术的实现，更是一种产品思维的体现。它把等待从一种“阻塞的负担”变成了一个“充满期待的进程”。当你看到自己亲手搭建的应用，能像真人对话一样，逐字逐句地给出智能回复时，那种成就感是巨大的。Spring AI Alibaba这套组合拳，极大地降低了Java开发者进入AI应用开发的门槛。从今天开始，试着把你的下一个Java项目加上一点AI的“流式”魔法吧。
更多推荐
9
0- 0
已为社区贡献6条内容
所有评论(0)