LLM 网关 SSE 流式转发实战：基于 SpringBoot 实现大模型输出透传与资源释放

m0_63990694

323人浏览 · 2026-11-02 19:21:21

m0_63990694 · 2026-11-02 19:21:21 发布

在大模型应用中，流式输出已经成为标准交互方式。相比一次性返回完整结果，SSE（Server-Sent Events）能够让用户实时感知模型生成过程，显著提升对话体验。

但在 Java 后端实现 LLM 网关的流式转发时，很多开发者会遇到几个典型问题：

把上游响应全部读取完成后再返回给前端，失去流式体验；
前端断开连接后，上游模型请求仍在继续，造成 Token 浪费；
不同大模型厂商的 SSE 事件格式不统一，适配成本高；
异步线程异常、超时、连接释放处理不当，导致服务稳定性问题。

本文基于 SpringBoot + OkHttp + ResponseBodyEmitter，实现一套生产可用的 SSE 流式转发方案，并重点说明连接生命周期管理、资源释放和多模型适配思路。

本文为系列第二篇。第一篇：《SpringBoot 实现大模型统一调度网关：OpenAI 与 Claude 协议适配》。

一、为什么 LLM 网关必须处理 SSE 流式输出

1. SSE 在大模型场景中的作用

SSE 是一种基于 HTTP 长连接的单向推送协议，服务端可以持续向客户端发送事件数据。在大模型对话中，模型输出通常不是瞬间完成的，而是逐步生成文本。通过 SSE，网关可以将模型生成的内容实时推送给前端。

一个标准的 SSE 响应格式如下：

data: {"content": "你好"}

data: {"content": "我是"}

data: {"content": "AI助手"}

data: [DONE]

其中：

data: 表示事件数据开头；
每段数据之间以 \n\n 分隔；
data: [DONE] 表示流结束。

2. 流式转发的核心难点

LLM 网关的流式转发，不只是“把数据读出来再写出去”这么简单。真正的难点在于生命周期管理：

请求开始时，如何建立长连接；
模型输出时，如何实时转发增量内容；
前端断开时，如何及时取消上游请求；
异常发生时，如何保证资源释放和错误传播；
不同模型之间，如何统一 SSE 事件格式。

因此，流式网关本质上是一个长连接代理系统，而不是简单的 HTTP 转发。

二、技术方案选型

实现流式输出，常见方案有以下几种：

方案	说明	是否适合 LLM 网关
普通返回 JSON	必须等待完整结果，不具备流式能力	不适合
WebSocket	双向通信，功能强，但协议较重	一般，大模型生态更偏向 SSE
Servlet 异步输出	可以实现流式响应，但代码较底层	可用
ResponseBodyEmitter	Spring 提供的异步响应工具，适合 SSE 场景	推荐

为什么选择 ResponseBodyEmitter

ResponseBodyEmitter 的优势主要有三点：

原生支持异步流式输出：可以在业务线程中持续写入数据；
连接生命周期回调完善：支持完成、超时、异常回调；
与 Spring Web 集成度高：不需要额外引入复杂组件。

因此，本文选择 ResponseBodyEmitter 作为 SSE 响应载体。

三、整体设计思路

我们的目标是：网关对外提供统一的 OpenAI 兼容 SSE 接口，内部适配不同厂商模型的流式协议。

整体流程如下：

客户端
  ↓
发送 stream=true 请求
  ↓
LLM 网关 Controller
  ↓
选择对应模型适配器（OpenAI / Claude）
  ↓
异步调用上游模型 SSE 接口
  ↓
逐行读取上游事件
  ↓
转换为统一 OpenAI SSE 格式
  ↓
实时写入 ResponseBodyEmitter
  ↓
客户端接收 data: 事件

关键设计原则：

统一对外协议：客户端只感知 OpenAI 兼容格式；
适配器隔离多模型差异：不同模型的 SSE 解析逻辑独立封装；
连接联动取消：客户端断开时，立即取消上游请求；
异步非阻塞处理：避免长连接占用 Tomcat 线程。

四、项目依赖

本文依赖与系列第一篇保持一致，核心使用 SpringBoot Web、OkHttp 和 Jackson。

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

<dependency>
    <groupId>com.squareup.okhttp3</groupId>
    <artifactId>okhttp</artifactId>
    <version>4.12.0</version>
</dependency>

<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
</dependency>

说明：

SpringBoot Web 提供 ResponseBodyEmitter；
OkHttp 负责调用上游大模型接口；
Jackson 负责请求参数和返回报文的 JSON 处理。

五、核心接口定义

为了让不同模型的流式处理保持统一，我们先定义一个模型适配器接口。

public interface ModelAdapter {

    /**
     * 非流式调用
     */
    String chat(ChatRequestDTO request) throws Exception;

    /**
     * 流式调用
     */
    void stream(ChatRequestDTO request, ResponseBodyEmitter emitter);
}

对应的工厂接口：

public interface ModelAdapterFactory {

    ModelAdapter getAdapter(String model);
}

这样，Controller 层只需要关心统一入口，不需要了解具体模型实现。

六、Controller 层实现

Controller 层负责接收客户端请求，并返回 ResponseBodyEmitter。

@RestController
@RequestMapping("/v1/chat/completions")
public class ChatCompletionController {

    private final ModelAdapterFactory adapterFactory;

    public ChatCompletionController(ModelAdapterFactory adapterFactory) {
        this.adapterFactory = adapterFactory;
    }

    @PostMapping
    public ResponseBodyEmitter chat(@RequestBody ChatRequestDTO request) {
        ModelAdapter adapter = adapterFactory.getAdapter(request.getModel());

        ResponseBodyEmitter emitter = new ResponseBodyEmitter(10 * 60 * 1000L);
        adapter.stream(request, emitter);

        return emitter;
    }
}

这里设置了 10 分钟超时，主要考虑长文本生成场景。实际项目中，超时时间应结合模型生成速度、业务场景和上游超时策略综合调整。

七、Claude SSE 流式适配器实现

Claude 的流式协议与 OpenAI 不同。它没有 [DONE] 结束标记，而是通过不同事件类型控制流生命周期。

1. Claude 常见 SSE 事件

事件类型	说明
`message_start`	消息开始
`content_block_delta`	内容增量，通常包含实际文本
`content_block_stop`	内容块结束
`message_stop`	整个消息结束

在适配时，我们只需要重点关注：

content_block_delta：提取增量文本；
message_stop：结束流。

2. Claude 适配器代码

@Component
public class ClaudeModelAdapter implements ModelAdapter {

    private final OkHttpClient okHttpClient;
    private final ObjectMapper objectMapper;

    public ClaudeModelAdapter(OkHttpClient okHttpClient, ObjectMapper objectMapper) {
        this.okHttpClient = okHttpClient;
        this.objectMapper = objectMapper;
    }

    @Override
    public void stream(ChatRequestDTO request, ResponseBodyEmitter emitter) {
        Executors.newCachedThreadPool().submit(() -> {
            try {
                Map<String, Object> body = buildClaudeRequestBody(request);

                Request httpRequest = new Request.Builder()
                        .url("[https://api.anthropic.com/v1/messages](https://api.anthropic.com/v1/messages)")
                        .header("x-api-key", System.getenv("CLAUDE_API_KEY"))
                        .header("anthropic-version", "2023-06-01")
                        .post(RequestBody.create(
                                objectMapper.writeValueAsString(body),
                                MediaType.parse("application/json")
                        ))
                        .build();

                Call call = okHttpClient.newCall(httpRequest);

                // 关键：客户端断开时，取消上游请求
                emitter.onCompletion(() -> call.cancel());
                emitter.onTimeout(() -> call.cancel());
                emitter.onError(throwable -> call.cancel());

                try (Response response = call.execute()) {
                    if (!response.isSuccessful() || response.body() == null) {
                        emitter.completeWithError(
                                new RuntimeException("Claude调用失败: " + response.code())
                        );
                        return;
                    }

                    BufferedReader reader = new BufferedReader(
                            new InputStreamReader(response.body().byteStream())
                    );

                    String line;
                    while ((line = reader.readLine()) != null) {
                        if (line.isBlank()) {
                            continue;
                        }

                        String openAiEvent = convertToOpenAiSse(line);
                        if (openAiEvent != null) {
                            emitter.send(openAiEvent);
                        }

                        if (line.contains("message_stop")) {
                            emitter.send("data: [DONE]\n\n");
                            emitter.complete();
                            break;
                        }
                    }
                }
            } catch (Exception e) {
                emitter.completeWithError(e);
            }
        });
    }

    @Override
    public String chat(ChatRequestDTO request) throws Exception {
        // 非流式逻辑，参考系列第一篇
        return null;
    }

    /**
     * 构造 Claude 请求体
     */
    private Map<String, Object> buildClaudeRequestBody(ChatRequestDTO request) {
        Map<String, Object> body = new HashMap<>();
        body.put("model", request.getModel());
        body.put("max_tokens", request.getMax_tokens() == null ? 1024 : request.getMax_tokens());
        body.put("stream", true);

        StringBuilder systemText = new StringBuilder();
        List<Map<String, Object>> messages = new ArrayList<>();

        for (ChatRequestDTO.ChatMessage message : request.getMessages()) {
            if ("system".equals(message.getRole())) {
                systemText.append(message.getContent());
            } else {
                Map<String, Object> msg = new HashMap<>();
                msg.put("role", message.getRole());
                msg.put("content", message.getContent());
                messages.add(msg);
            }
        }

        if (systemText.length() > 0) {
            body.put("system", systemText.toString());
        }
        body.put("messages", messages);

        return body;
    }

    /**
     * 将 Claude SSE 事件转换为 OpenAI 兼容格式
     */
    private String convertToOpenAiSse(String line) {
        // 实际项目中应解析 JSON，提取 content_block_delta 中的文本
        // 这里为结构示意，真实实现需要根据 Claude 返回字段精确映射
        if (line.contains("content_block_delta")) {
            return "data: {\"choices\":[{\"delta\":{\"content\":\"文本增量\"}}]}\n\n";
        }
        return null;
    }
}

3. 代码说明

这段代码的核心逻辑是：

在异步线程中调用 Claude SSE 接口；
注册 emitter 回调，保证连接断开时取消上游请求；
逐行读取 Claude 事件；
将 content_block_delta 转换为 OpenAI 的 delta 格式；
遇到 message_stop 时，发送 [DONE] 并结束流。

八、OpenAI 兼容模型的流式处理

对于 OpenAI 原生接口，以及兼容 OpenAI 协议的国产模型，流式处理会简单很多。

1. OpenAI SSE 特点

事件格式统一为 data: {...}；
增量内容在 choices[0].delta.content 中；
结束标记为 data: [DONE]。

2. OpenAI 适配器实现

@Component
public class OpenAiModelAdapter implements ModelAdapter {

    private final OkHttpClient okHttpClient;
    private final ObjectMapper objectMapper;

    public OpenAiModelAdapter(OkHttpClient okHttpClient, ObjectMapper objectMapper) {
        this.okHttpClient = okHttpClient;
        this.objectMapper = objectMapper;
    }

    @Override
    public void stream(ChatRequestDTO request, ResponseBodyEmitter emitter) {
        Executors.newCachedThreadPool().submit(() -> {
            try {
                Request httpRequest = new Request.Builder()
                        .url("[https://api.openai.com/v1/chat/completions](https://api.openai.com/v1/chat/completions)")
                        .header("Authorization", "Bearer " + System.getenv("OPENAI_API_KEY"))
                        .post(RequestBody.create(
                                objectMapper.writeValueAsString(request),
                                MediaType.parse("application/json")
                        ))
                        .build();

                Call call = okHttpClient.newCall(httpRequest);

                emitter.onCompletion(() -> call.cancel());
                emitter.onTimeout(() -> call.cancel());
                emitter.onError(throwable -> call.cancel());

                try (Response response = call.execute()) {
                    if (!response.isSuccessful() || response.body() == null) {
                        emitter.completeWithError(
                                new RuntimeException("OpenAI调用失败: " + response.code())
                        );
                        return;
                    }

                    BufferedReader reader = new BufferedReader(
                            new InputStreamReader(response.body().byteStream())
                    );

                    String line;
                    while ((line = reader.readLine()) != null) {
                        if (line.isBlank()) {
                            continue;
                        }
                        emitter.send(line + "\n\n");

                        if ("data: [DONE]".equals(line)) {
                            emitter.complete();
                            break;
                        }
                    }
                }
            } catch (Exception e) {
                emitter.completeWithError(e);
            }
        });
    }

    @Override
    public String chat(ChatRequestDTO request) throws Exception {
        // 非流式逻辑
        return null;
    }
}

对于 OpenAI 兼容模型，大部分情况下可以直接透传 data: 行，不需要复杂的事件转换。

九、连接断开与资源释放

1. 为什么必须取消上游请求

如果客户端主动关闭页面、取消请求或网络中断，而网关没有及时取消上游模型调用，模型可能继续生成内容，导致：

Token 费用浪费；
上游连接占用；
无效日志增多；
服务端资源泄漏。

因此，必须建立连接联动机制。

2. 连接取消逻辑

emitter.onCompletion(() -> call.cancel());
emitter.onTimeout(() -> call.cancel());
emitter.onError(throwable -> call.cancel());

这三个回调分别表示：

onCompletion：正常完成；
onTimeout：超时；
onError：发生异常。

无论哪种情况，都应取消上游 Call。

3. OkHttp 的 cancel 是否可靠

OkHttp 的 call.cancel() 会尝试中断正在执行的请求。对于正在读取响应体的长连接，取消后会关闭底层连接，从而终止数据读取。

但需要注意：

取消操作不保证上游模型立即停止生成；
但可以保证网关不再继续读取和转发数据；
这已经能避免大部分无效 Token 消耗和资源占用。

十、SSE 格式规范与常见坑

1. 必须使用 `\n\n` 结束事件

SSE 协议要求每个事件以 \n\n 结尾。如果只写一个换行或不加换行，前端可能无法正确识别事件边界。

正确写法：

emitter.send("data: " + json + "\n\n");

2. 不能缓存完整响应

流式处理的关键是“来一段、处理一段、发一段”。

错误做法：

把所有 chunk 读到内存中；
拼接成完整字符串后再返回。

正确做法：

BufferedReader 逐行读取；
解析一行；
转换一行；
立即发送一行。

3. 异常不能直接抛出

在异步线程中，异常不能像同步方法那样直接抛出，否则可能导致线程终止或容器异常。

统一使用：

emitter.completeWithError(e);

4. 超时时间要合理

大模型生成速度较慢，特别是长文本、复杂推理场景。如果超时时间设置过短，会导致流被意外中断。

建议：

普通对话：30 秒到 2 分钟；
长文本生成：5 到 10 分钟；
同时配置 OkHttp 读取超时和 emitter 超时。

十一、生产环境优化建议

1. 线程池隔离

流式请求占用时间较长，建议将 SSE 处理逻辑与普通业务接口分离，使用独立线程池。

@Bean
public ExecutorService sseExecutor() {
    return new ThreadPoolExecutor(
            4,
            16,
            60L, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(200),
            new ThreadFactoryBuilder().setNameFormat("sse-executor-%d").build()
    );
}

2. OkHttp 连接池配置

长连接较多时，需要合理配置 OkHttp 连接池，避免连接耗尽。

@Bean
public OkHttpClient okHttpClient() {
    return new OkHttpClient.Builder()
            .connectionPool(new ConnectionPool(32, 5, TimeUnit.MINUTES))
            .readTimeout(10, TimeUnit.MINUTES)
            .build();
}

3. 增加熔断与降级

当上游模型连续超时或大量报错时，网关应具备熔断能力，避免无效请求持续打到上游。

可结合：

Redis 计数失败次数；
滑动窗口统计错误率；
超过阈值后暂时拒绝或切换备用模型。

4. 日志与监控

建议记录以下指标：

请求开始/结束时间；
模型名称；
输入 Token 数；
输出 Token 数；
首 Token 延迟；
平均生成速度；
断开原因；
异常类型。

这些指标不仅用于排查问题，也是后续成本核算和毛利统计的基础。

十二、总结

我在大模型统一调度网关项目中负责 SSE 流式转发模块。网关对外提供统一的 OpenAI 兼容接口，内部适配 OpenAI 和 Claude 等模型的流式协议。我使用 SpringBoot 的 ResponseBodyEmitter 处理异步长连接，通过 OkHttp 调用上游模型，并逐行解析 SSE 事件，转换成统一格式后实时推送给前端。

为避免前端断开后上游请求继续消耗 Token，我在 ResponseBodyEmitter 的完成、超时和异常回调中统一取消 OkHttp 请求，实现连接生命周期联动。同时，针对不同模型的事件格式做了适配器封装，Claude 主要处理 content_block_delta 和 message_stop 事件，OpenAI 模型则直接透传标准 SSE 数据。

此外，我还做了线程池隔离、连接池优化、超时控制和异常处理，提升流式接口的稳定性。

总结

LLM 网关的 SSE 流式转发，核心并不在于“转发”本身，而在于长连接生命周期管理。

一个生产可用的流式网关，至少需要解决四件事：

实时转发：模型生成内容实时推送到前端；
协议适配：统一不同模型的 SSE 事件格式；
资源释放：客户端断开时及时取消上游请求；
稳定性保障：超时、异常、线程池、连接池和监控要配套。

本文实现的方案，已经具备一个小型企业级 LLM 网关流式模块的基础骨架。在此之上，可以继续叠加限流、鉴权、Token 计费、成本统计、熔断降级等能力，形成完整的大模型统一接入平台。

北京朝阳AI社区

更多推荐

程序员转型抉择：转 Agent 开发还是转全栈？一份基于招聘市场的冷静分析

北京朝阳AI社区

10 道 AI 安全工程师面试真题！小白程序员吃透这套实战技巧，Offer 到手

北京朝阳AI社区

《大模型原理通俗讲》系列 13｜RAG 通俗讲：给大模型外接一本书

RAG就像给大模型外接一本“参考书”：当用户提问时，系统先从资料库中检索最相关的段落，再将这些内容作为“提示”输入模型，让其据此回答。相比直接编造或塞入大量文本，RAG能精准调用私有、动态更新的资料，减少幻觉，且无需重训练模型。关键在于用向量“按意思”找资料，而非死扣关键词，使模型真正“懂”上下文。