Spring Boot 对接大模型 API 实战：SSE 流式输出、超时重试与优雅降级

分类：AI 与智能体 · 作者：管理员 · 发布于 2026-10-04 · 阅读 0 · 标签：Spring Boot / SSE / 大模型 / 流式输出 / 重试降级

一、痛点：AI 接口为什么不能同步调用

业务侧要给用户提供 AI 智能助手，比如电商小程序的商品导购、售后答疑。大模型接口有三个典型特征：响应慢（首字要几秒，全文可能几十秒）、不稳定（限流 429、网关超时、5xx 随时出现）、按 token 计费（重试可能重复扣费）。

如果按普通 HTTP 同步调用的方式处理，Servlet 线程会被占住几十秒，Tomcat 默认 200 个线程很快耗尽，其他普通接口全部排队；前端用户也只能盯着空白页面干等。结论只有一个：必须流式输出，把「等待完整结果」变成「持续接收增量内容」。

二、方案设计：为什么选 SSE

实时通道常见三种：WebSocket、SSE（Server-Sent Events）、轮询。

WebSocket：双向、二进制友好，但要处理心跳、断线重连、协议升级，小程序端也要引第三方库，复杂度最高；
轮询：定时拉取，实时性差，流量浪费严重；
SSE：基于普通 HTTP，服务端单向推送文本，浏览器 EventSource 原生支持，断了自动重连，对一个「后端逐字转发」的场景几乎零成本。

典型架构：小程序前端 → 中台网关（Spring Boot）→ 上游大模型 API。网关层承担四件事：透传流式数据、识别上游错误、控制超时、失败降级。

三、代码实现

3.1 依赖与基础配置

<dependency>
    <groupId>com.squareup.okhttp3</groupId>
    <artifactId>okhttp</artifactId>
    <version>4.12.0</version>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>

app:
  llm:
    base-url: https://api.example.com/v1/chat/completions
    connect-timeout: 3s        # 建连超时，快速失败
    read-timeout: 60s          # 单次读超时，覆盖首字延迟
    idle-timeout: 90s          # 整体兜底：超过 90 秒强制断开
    max-retries: 2             # 可重试错误最多重试次数
    retry-backoff-base: 500ms  # 指数退避基数

3.2 用 OkHttp 接收上游流

上游大模型接口开启 stream=true 后，会持续返回 data: {...} 行。用 OkHttp 的 EventSource 按行解析，天然规避了 TCP 分包导致的中文乱码问题（不要自己拼字节再 decode）。

OkHttpClient client = new OkHttpClient.Builder()
        .connectTimeout(3, TimeUnit.SECONDS)
        .readTimeout(60, TimeUnit.SECONDS)
        .build();

Request request = new Request.Builder()
        .url(baseUrl)
        .header("Authorization", "Bearer " + apiKey)
        .header("Content-Type", "application/json")
        .post(RequestBody.create(
                MediaType.parse("application/json; charset=utf-8"),
                jsonPayload))
        .build();

EventSource.Factory factory = EventSources.createFactory(client);
factory.newEventSource(request, new EventSourceListener() {
    @Override
    public void onEvent(EventSource es, String id, String type, String data) {
        // type 为 error 时是上游错误事件，与内容事件严格区分
        if ("error".equals(type)) {
            handleUpstreamError(data);
            return;
        }
        if ("[DONE]".equals(data)) {
            onDone.accept(null);
            return;
        }
        // 解析增量 delta，转发给前端
        onChunk.accept(parseDelta(data));
    }

    @Override
    public void onFailure(EventSource es, Throwable t, Response response) {
        // 网络异常或上游非 2xx：交给重试/降级逻辑
        onError.accept(t, response);
    }
});

3.3 SseEmitter 向小程序前端转发

@GetMapping("/ai/chat")
public SseEmitter chat(@RequestParam String question) {
    SseEmitter emitter = new SseEmitter(90_000L); // 与 idle-timeout 对齐
    executor.submit(() -> invokeLlm(question, emitter)); // 固定线程池，禁止 new Thread
    return emitter;
}

private void invokeLlm(String question, SseEmitter emitter) {
    try {
        llmClient.streamChat(question,
            chunk -> emitter.send(SseEmitter.event().data(chunk)),
            () -> emitter.complete(),
            (err, resp) -> {
                if (err != null) emitter.completeWithError(err);
                else emitter.complete(); // 上游错误已转成友好文案发送
            });
    } catch (Exception e) {
        emitter.completeWithError(e);
    }
}

3.4 重试与降级封装

重试不是无脑重来，必须区分错误类型：

网络异常、5xx、429（带 Retry-After）：可重试，指数退避；
4xx 业务错误（参数错、鉴权失败）：重试无用，直接失败；
流已经开始出字后断开：不自动重试——重试会导致内容重复、重复计费，应提示用户重新发起；
重试请求必须带 幂等键（如 Idempotency-Key: 请求号），防止上游重复生成。

private void invokeWithRetry(String question, Consumer<String> onChunk) {
    int attempt = 0;
    long backoff = retryBackoffBase;
    while (attempt <= maxRetries) {
        try {
            streamOnce(question, onChunk);
            return; // 正常完成
        } catch (RetryableException e) {
            if (attempt == maxRetries) break;
            sleep(backoff);
            backoff *= 2;
        } catch (NonRetryableException e) {
            throw e; // 4xx / 已出字后断开，直接失败
        }
        attempt++;
    }
    // 全部失败：降级兜底
    onChunk.accept("抱歉，服务暂时繁忙，请稍后再试。");
    alertService.send("LLM 调用连续失败", question);
}

四、踩坑记录

SseEmitter 不显式 complete 会一直 pending：前端永远转圈。必须在 onCompletion / onTimeout / onError 里都走到 complete，否则连接泄漏；超时回调里要 completeWithError 并记日志。
上游限流文案被当成正文透传：上游 429 时错误信息也走 SSE 通道，若不区分事件类型，用户会看到一串「当前请求过多，请稍后再试」的机器文案甚至 JSON 原文。正确做法是把上游错误解析进 error 事件或 HTTP 状态码，转成统一友好文案，并把原始错误写入服务端日志与告警，绝不透传给用户。
连接超时、读超时、整体兜底是三个维度：connectTimeout 只管建连；readTimeout 只保证「每次读到数据」；还要一个整体 idle 计时，避免「上游静默挂死、连接既不报错也不出字」的情况。三者缺一不可。
每次请求 new OkHttpClient：连接池被浪费，还容易触发 TIME_WAIT 堆积。客户端必须是单例，EventSource 用完 close 释放连接。
异步线程池没上限：AI 请求量大时线程池打满，普通接口跟着遭殃。用固定线程池 + 有界队列，拒绝策略选 CallerRuns 或直接降级返回。

五、最佳实践清单

抽象 LLMClient 接口：把「流式/非流式、重试、超时」封装在实现里，业务层只依赖接口；多个模型供应商可路由切换，一个挂了换另一个；
配置化：超时、重试次数、退避基数、降级文案全部放 @ConfigurationProperties，上线后不重启就能调参；
监控三指标：TTFT（首字延迟）、token 用量、429/5xx 错误率。首字延迟是用户体验的命门，必须单独埋点；
双通道兜底：流式失败且不可重试时，回退到非流式接口拿完整结果，仍失败再降级文案；
限流与熔断：网关层对 AI 接口单独限流（AI 调用贵且慢，不能和普通接口共用配额），上游连续失败达到阈值时熔断一段时间，避免雪崩。

SSE 流式是 AI 应用后端的「基本功」，但把超时、重试、降级、连接管理做扎实，才能让用户在高峰期也稳定地「看到字在出来」。这套思路同样适用于语音转写、图片生成等一切慢接口的异步化改造。