Spring AI企业级应用实战（3）：流式输出、SSE、背压、取消与异常恢复

原创 21天前 1791

Spring AI企业级应用实战（3）：流式输出、SSE、背压、取消与异常恢复

文章摘要

上一篇完成了ChatClient、Advisor与Prompt Template统一调用层。本篇继续构建生产级流式能力：使用Spring WebFlux和Spring AI返回结构化SSE事件，实现首Token统计、Token合并、背压保护、用户取消、心跳、超时、部分结果保存、错误事件和Nginx部署配置。文章给出完整的请求模型、任务状态、Service、Controller与前端读取代码，解决“本地能流式、上线后一次性返回”的常见问题。

一、本篇目标

最终调用链：

浏览器
→ POST流式请求
→ Spring WebFlux Controller
→ StreamingAiService
→ Enterprise ChatClient
→ Spring AI ChatModel
→ SSE事件
→ 浏览器逐段渲染

需要支持：

结构化SSE；
任务ID；
开始、增量、错误和结束事件；
首Token耗时；
用户取消；
慢客户端保护；
空闲心跳；
总超时；
部分结果保存；
Nginx无缓冲部署。

二、为什么不直接返回Flux

最小代码：

return chatClient.prompt()
        .user(message)
        .stream()
        .content();

适合Demo，但前端无法区分：

普通文本
工具事件
错误
Usage
任务结束

生产环境建议事件协议：

{
  "taskId": "task-123",
  "sequence": 12,
  "type": "delta",
  "data": {
    "content": "正在"
  }
}

三、项目依赖

            org.springframework.ai
            spring-ai-bom
            2.0.0
            pom
            import






        org.springframework.ai
        spring-ai-starter-model-deepseek



        org.springframework.boot
        spring-boot-starter-webflux



        org.springframework.boot
        spring-boot-starter-actuator



        org.springframework.boot
        spring-boot-starter-validation

Spring WebFlux基于Reactive Streams，支持非阻塞处理和背压。

四、定义请求

public record AiStreamRequest(
        @NotBlank
        String message,
        String conversationId
) {
}

请求上下文：

public record AiRequestContext(
        String requestId,
        String taskId,
        String userId,
        String tenantId
) {
}

生产环境从认证上下文读取用户和租户，不直接信任客户端请求字段。

五、定义事件

public record AiStreamEvent(
        String taskId,
        long sequence,
        String type,
        Map data,
        long timestamp
) {
}

事件类型：

public final class StreamEventType {

    public static final String START = "start";
    public static final String DELTA = "delta";
    public static final String HEARTBEAT = "heartbeat";
    public static final String USAGE = "usage";
    public static final String ERROR = "error";
    public static final String DONE = "done";

    private StreamEventType() {
    }
}

六、任务状态

public enum StreamTaskStatus {
    CREATED,
    STREAMING,
    CANCEL_REQUESTED,
    CANCELLED,
    COMPLETED,
    FAILED,
    TIMED_OUT
}

任务对象：

public class StreamTask {

    private final String taskId;
    private final String userId;
    private final String tenantId;
    private final Instant createdAt;
    private final AtomicReference
            status;

    public StreamTask(
            String taskId,
            String userId,
            String tenantId
    ) {
        this.taskId = taskId;
        this.userId = userId;
        this.tenantId = tenantId;
        this.createdAt = Instant.now();
        this.status = new AtomicReference(
                StreamTaskStatus.CREATED
        );
    }

    public void markStreaming() {
        status.set(StreamTaskStatus.STREAMING);
    }

    public void requestCancel() {
        status.set(
                StreamTaskStatus.CANCEL_REQUESTED
        );
    }

    public boolean isCancelRequested() {
        return status.get()
                == StreamTaskStatus.CANCEL_REQUESTED;
    }
}

七、任务仓库

演示使用内存：

@Component
public class InMemoryStreamTaskRepository {

    private final ConcurrentMap
            tasks = new ConcurrentHashMap();

    public StreamTask create(
            String userId,
            String tenantId
    ) {
        String taskId = UUID.randomUUID()
                .toString();

        StreamTask task = new StreamTask(
                taskId,
                userId,
                tenantId
        );

        tasks.put(taskId, task);
        return task;
    }

    public StreamTask require(String taskId) {
        StreamTask task = tasks.get(taskId);

        if (task == null) {
            throw new IllegalArgumentException(
                    "任务不存在"
            );
        }

        return task;
    }
}

多实例生产部署应使用Redis或数据库保存可共享状态。

八、事件工厂

@Component
public class AiStreamEventFactory {

    public AiStreamEvent create(
            String taskId,
            AtomicLong sequence,
            String type,
            Map data
    ) {
        return new AiStreamEvent(
                taskId,
                sequence.incrementAndGet(),
                type,
                data,
                System.currentTimeMillis()
        );
    }
}

九、核心Streaming Service

@Service
public class StreamingAiService {

    private final ChatClient chatClient;
    private final InMemoryStreamTaskRepository
            taskRepository;
    private final AiStreamEventFactory eventFactory;

    public StreamingAiService(
            @Qualifier("enterpriseChatClient")
            ChatClient chatClient,
            InMemoryStreamTaskRepository taskRepository,
            AiStreamEventFactory eventFactory
    ) {
        this.chatClient = chatClient;
        this.taskRepository = taskRepository;
        this.eventFactory = eventFactory;
    }

    public StreamResult stream(
            AiStreamRequest request,
            AuthenticatedUser user
    ) {
        StreamTask task = taskRepository.create(
                user.userId(),
                user.tenantId()
        );

        AtomicLong sequence = new AtomicLong();
        AtomicBoolean firstToken =
                new AtomicBoolean(true);
        long startedAt = System.nanoTime();
        StringBuilder answerBuffer =
                new StringBuilder();

        AiStreamEvent startEvent =
                eventFactory.create(
                        task.taskId(),
                        sequence,
                        StreamEventType.START,
                        Map.of(
                                "requestId",
                                UUID.randomUUID()
                                        .toString()
                        )
                );

        Flux modelEvents =
                chatClient.prompt()
                        .advisors(spec -> spec
                                .param(
                                        "tenantId",
                                        user.tenantId()
                                )
                                .param(
                                        "conversationId",
                                        request.conversationId()
                                )
                        )
                        .user(request.message())
                        .stream()
                        .content()
                        .takeUntilOther(
                            cancelSignal(task)
                        )
                        .bufferTimeout(
                                12,
                                Duration.ofMillis(40)
                        )
                        .filter(parts ->
                                !parts.isEmpty()
                        )
                        .map(parts ->
                                String.join("", parts)
                        )
                        .map(content -> {
                            if (
                                firstToken.compareAndSet(
                                        true,
                                        false
                                )
                            ) {
                                long firstTokenMs =
                                        (System.nanoTime()
                                         - startedAt)
                                        / 1_000_000;

                                metrics.recordFirstToken(
                                        firstTokenMs
                                );
                            }

                            answerBuffer.append(content);

                            return eventFactory.create(
                                    task.taskId(),
                                    sequence,
                                    StreamEventType.DELTA,
                                    Map.of(
                                            "content",
                                            content
                                    )
                            );
                        });

        Flux stream =
                Flux.concat(
                        Mono.just(startEvent),
                        modelEvents
                )
                .timeout(Duration.ofMinutes(5))
                .onErrorResume(
                        TimeoutException.class,
                        error -> Flux.just(
                            errorEvent(
                                    task,
                                    sequence,
                                    "TASK_TIMEOUT",
                                    "生成超时"
                            )
                        )
                )
                .onErrorResume(error -> Flux.just(
                    errorEvent(
                            task,
                            sequence,
                            "STREAM_FAILED",
                            "生成过程中发生错误"
                    )
                ))
                .concatWith(
                        Mono.fromSupplier(() ->
                            doneEvent(
                                    task,
                                    sequence
                            )
                        )
                )
                .doOnSubscribe(subscription ->
                        task.markStreaming()
                )
                .doOnCancel(() ->
                        savePartial(
                                task,
                                answerBuffer.toString()
                        )
                )
                .doOnComplete(() ->
                        saveCompleted(
                                task,
                                answerBuffer.toString()
                        )
                )
                .doFinally(signal ->
                        recordFinalSignal(
                                task,
                                signal
                        )
                );

        return new StreamResult(
                task.taskId(),
                stream
        );
    }
}

示例中使用了：

bufferTimeout

把极小Token合并后发送，减少网络和DOM更新开销。

十、取消信号

private Publisher cancelSignal(
        StreamTask task
) {
    return Flux.interval(
            Duration.ofMillis(100)
        )
        .filter(index ->
                task.isCancelRequested()
        )
        .next();
}

这是便于理解的演示实现。

生产环境更适合：

Reactor Sink；
Redis Pub/Sub；
消息总线；
Provider显式取消接口。

使用轮询会增加不必要的调度。

十一、取消接口

@PostMapping("/tasks/{taskId}/cancel")
public ResponseEntity cancel(
        @AuthenticationPrincipal
        AuthenticatedUser user,
        @PathVariable String taskId
) {
    StreamTask task =
            taskRepository.require(taskId);

    if (!task.userId().equals(user.userId())) {
        throw new AccessDeniedException(
                "无权取消该任务"
        );
    }

    task.requestCancel();

    return ResponseEntity.accepted().build();
}

取消必须校验任务所有者和租户。

十二、Controller返回SSE

@PostMapping(
    value = "/stream",
    produces = MediaType.TEXT_EVENT_STREAM_VALUE
)
public ResponseEntity>
> stream(
        @AuthenticationPrincipal
        AuthenticatedUser user,
        @Valid @RequestBody
        AiStreamRequest request
) {
    StreamResult result =
            streamingAiService.stream(
                    request,
                    user
            );

    Flux> body =
            result.events()
                    .map(event ->
                        ServerSentEvent
                                .builder()
                                .id(
                                    Long.toString(
                                        event.sequence()
                                    )
                                )
                                .event(event.type())
                                .data(event)
                                .build()
                    );

    return ResponseEntity.ok()
            .header(
                    HttpHeaders.CACHE_CONTROL,
                    "no-cache, no-transform"
            )
            .header(
                    "X-Accel-Buffering",
                    "no"
            )
            .header(
                    "X-AI-Task-Id",
                    result.taskId()
            )
            .body(body);
}

十三、心跳

模型调用工具时可能较长时间无Token。

Flux heartbeat =
        Flux.interval(Duration.ofSeconds(15))
                .map(index ->
                    eventFactory.create(
                            task.taskId(),
                            sequence,
                            StreamEventType.HEARTBEAT,
                            Map.of()
                    )
                );

将心跳与业务流合并时，要在任务完成后停止心跳：

Flux.merge(
    businessStream,
    heartbeat.takeUntilOther(
        businessStream.ignoreElements()
    )
)

实际实现应避免对冷流重复订阅，可使用publish或共享完成信号。

十四、背压与缓冲

WebFlux支持Reactive Streams背压，但模型SDK、网络和前端消费速度不一定完全一致。

保护：

.onBackpressureBuffer(
        256,
        dropped -> log.warn(
                "stream overflow taskId={}",
                task.taskId()
        ),
        BufferOverflowStrategy.ERROR
)

不要设置无限缓冲。

文本场景可使用：

小Token合并
＋有限缓冲
＋慢客户端超时

十五、部分结果保存

正常完成：

status = COMPLETED
partial = false

用户取消：

status = CANCELLED
partial = true

异常中断：

status = FAILED
partial = true

部分结果不能进入完整答案缓存，也不能作为后续Memory中的权威回答。

十六、前端读取

const controller = new AbortController();

async function startChat(message) {
  const response = await fetch("/api/ai/stream", {
    method: "POST",
    headers: {
      "Content-Type": "application/json"
    },
    body: JSON.stringify({
      message,
      conversationId
    }),
    signal: controller.signal
  });

  const taskId = response.headers.get(
    "X-AI-Task-Id"
  );

  const reader = response.body.getReader();
  const decoder = new TextDecoder();
  let buffer = "";

  while (true) {
    const { value, done } = await reader.read();
    if (done) break;

    buffer += decoder.decode(value, {
      stream: true
    });

    buffer = parseSseFrames(
      buffer,
      event => handleEvent(event)
    );
  }

  return taskId;
}

停止：

async function stop(taskId) {
  controller.abort();

  await fetch(
    `/api/ai/tasks/${taskId}/cancel`,
    { method: "POST" }
  );
}

十七、Nginx配置

location /api/ai/stream {
    proxy_pass http://spring_ai_backend;
    proxy_http_version 1.1;

    proxy_buffering off;
    proxy_cache off;
    gzip off;

    proxy_read_timeout 600s;
    proxy_send_timeout 600s;

    proxy_set_header Host $host;
    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;

    add_header X-Accel-Buffering no;
}

如果前面还有CDN或API Gateway，也要确认其支持SSE和长连接。

十八、错误事件

private AiStreamEvent errorEvent(
        StreamTask task,
        AtomicLong sequence,
        String code,
        String message
) {
    return eventFactory.create(
            task.taskId(),
            sequence,
            StreamEventType.ERROR,
            Map.of(
                    "code",
                    code,
                    "message",
                    message
            )
    );
}

不要把Provider异常堆栈直接返回用户。

十九、关键监控指标

stream_requests_total
stream_active_connections
first_token_latency_ms
stream_duration_ms
stream_cancel_rate
stream_timeout_rate
stream_error_rate
buffer_overflow_count
partial_answer_count
provider_tokens_after_cancel

重点关注：

P95首Token耗时；
用户取消率；
取消传播耗时；
超时率；
代理层断连率。

二十、常见错误

collectList后返回
在map中使用JDBC
Nginx开启buffering
前端使用response.text()
取消只隐藏页面
无限onBackpressureBuffer
每个Token写数据库
心跳无法停止
错误后仍发送done成功事件

二十一、完整生产链路

用户提交请求
→ 创建taskId
→ 权限与配额校验
→ ChatClient流式调用
→ Token合并
→ SSE事件
→ 首Token与耗时监控
→ 用户取消或正常完成
→ 保存完整或部分结果
→ 记录Token、费用与Trace

总结

生产级Spring AI流式接口，不只是把：

.stream().content()

返回给Controller。

还必须处理：

事件协议
背压
取消
心跳
超时
部分结果
代理缓冲
错误恢复
监控

完成这些能力后，流式输出才能从演示功能变成稳定的企业AI交互基础设施。

下一篇将继续实现：

Spring AI企业级应用实战（4）：Chat Memory、多租户会话与长期记忆治理。

延伸阅读

如果你正在关注企业级 AI 应用、Agent、RAG、MCP 与大模型工程化落地，欢迎访问 智元界：

https://www.zyentor.com/

智元界将持续分享可运行的技术实战、架构设计、问题排查与企业应用案例。

📚 Spring AI企业级应用实战

← 上一篇

Spring AI企业级应用实战（2）：ChatClient、Advisor与Prompt Template构建统一调用层

Spring AI企业级应用实战（4）：Chat Memory、会话隔离、持久化与上下文压缩

认真做体验路线图 2026/07/19 10:16:39

确实，直接交付结果对准确率和成本要求太高，我们试过类似方案，用户纠错成本反而更大。

小白LinuxLab 2026/07/18 08:03:19

确实，直接交付结果听着美好，实际容错和成本太难平衡了。

Spring AI企业级应用实战（3）：流式输出、SSE、背压、取消与异常恢复

Spring AI企业级应用实战（3）：流式输出、SSE、背压、取消与异常恢复

文章摘要

一、本篇目标

二、为什么不直接返回Flux

三、项目依赖

四、定义请求

五、定义事件

六、任务状态

七、任务仓库

八、事件工厂

九、核心Streaming Service

十、取消信号

十一、取消接口

十二、Controller返回SSE

十三、心跳

十四、背压与缓冲

十五、部分结果保存

十六、前端读取

十七、Nginx配置

十八、错误事件

十九、关键监控指标

二十、常见错误

二十一、完整生产链路

总结

延伸阅读

评论 2

📖 关联推荐

精选推荐