最近在做一个 AI 流式输出的服务端对接需求很典型上游给的是 SSE 流式接口一个接一个地出 token但业务侧是传统的 Spring Boot 项目既要消费这个流又要实时转发给前端页面。最后落地就是用 WebClient 拉取 SSE 流经过几轮改造稳定跑起来了。如果你正在用 Spring Boot 对接实时日志、行情推送、AI 流式结果这类接口这篇实战复盘应该能帮你省不少时间尤其是那些文档里不会写明白的参数和调度细节我会一并摊开讲。1. 先搞清楚场景SSE 长连接和 WebClient 为什么这么搭1.1 SSE 协议本质与定位SSE 全称 Server-Sent Events是一个建立在 HTTP 之上的服务端单向推送协议。服务端用Content-Type: text/event-stream维持一个长响应按事件格式持续写出数据。它的数据是纯文本按行组织的data:行放业务内容event:行放事件类型id:行放事件唯一标识字段之间用空行分隔以冒号开头的行属于注释通常被用来做心跳。这个协议最大的价值是“省事”。链路两端只需要 HTTP 栈不需要 WebSocket 那样的握手帧和状态机经过负载均衡、网关、代理时兼容性成本低很多。对比轮询它省掉大量无效请求对比 WebSocket它没有双向通道服务端推什么前端接收什么协议本身天然比复杂的长连接协议更不容易出错。做服务端集成的时候“简单”本身就是一种很大的优势排查链路时你能肉眼分析出每一帧数据的含义。1.2 消费侧为什么用 WebClientJava 生态里能消费 SSE 的客户端不止一个但真正适合这个场景的是 WebClient。RestTemplate虽然在 Spring 里很常用但对流式响应的支持非常别扭它本质上是同步阻塞模型拿到一个无限输出流时很难优雅地按条处理OkHttp能发流式请求可事件格式解析、断线重连、背压控制都要自己手写成本不低。WebClient 是响应式客户端天生为流式场景设计Spring 官方还内置了text/event-stream的解析器底层默认跑 Reactor Netty背压链路完整。消费 SSE 这件事几乎可以算是它最典型的用法。另外很重要的一点是 WebClient 对长连接生命周期的管理。SSE 流的持续时间可能几分钟也可能几小时中途的数据量、发送频率都是动态的。响应式编程里Flux 天然支持按需拉取、按量控制、随时取消写出来的代码就是在描述数据流的变换过程不需要你自己维护线程池和状态标志。这一点对后端服务尤其关键因为线上并发连接一多资源管理就决定了系统能不能在高峰期活下来。1.3 什么场景适合这套组合结合我接触过的项目这几类需求都适合用“SSE WebClient”AI 大模型对话场景的 token 流式返回运维平台的实时日志推送行情和监控指标这类高频小数据流以及需要把第三方流式消息源接入自有业务系统的中间层。判断标准很简单数据是服务端单向往客户端推、连接持续时间长、对实时性有要求基本就是 SSE 的地盘。反过来如果业务里客户端需要频繁往服务端推消息那就是 WebSocket 的范畴了别用 SSE 硬扛。2. 工程落地依赖引入与连接参数配置2.1 依赖与 Boot 版本注意点先说依赖怎么加。如果你原来的项目是spring-boot-starter-web不需要为了用 WebClient 重写整个应用直接加一个spring-boot-starter-webflux就行dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-webflux/artifactId /dependency很多人担心 web 和 webflux 两个 starter 共存会冲突实际上在 Spring Boot 2.x 和 3.x 里MVC 应用依然以 Servlet 容器启动WebClient 只是作为响应式 HTTP 客户端组件被使用两者可以平安共存。我见过一些人为了“避免冲突”去手动排除依赖结果反而把自动配置弄出问题得不偿失。版本上也要注意。Spring Boot 2.3 之前 WebClient 的 API 有过一些调整如果你还在用老的 2.1、2.2建议先把 Boot 升上去再套用本文的代码。我现在用 Boot 3.x 配合 WebFlux 6代码风格已经比较稳定新项目直接照这个基调走。2.2 连接超时与响应超时的平衡配置 WebClient最关键的往往不是请求怎么写而是超时参数。SSE 长连接里有两个超时维度很多人会混在一起一个是 TCP 建连超时connectTimeout这个按常规思路设置3 到 5 秒比较合理别让一个不可达的地址拖住整体链路。另一个是对端数据静默时的读超时responseTimeout这才是长连接的命门。如果你按照普通 HTTP 请求的习惯设成 5 秒、10 秒服务端发心跳稍微慢一点连接就被你主动掐断了如果设成 0某些网关场景又会因为没有兜底而导致连接泄漏。我的习惯是对接方明确有心跳就设置 10 到 30 分钟没有心跳且对方稳定性一般至少给 5 分钟并在重试逻辑里覆盖超时场景。一个可用的底座配置长这样HttpClient httpClient HttpClient.create() .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000) .responseTimeout(Duration.ofMinutes(10)); ReactorClientHttpConnector connector new ReactorClientHttpConnector(httpClient); WebClient webClient WebClient.builder() .baseUrl(http://ai.example.com) .clientConnector(connector) .build();responseTimeout在 Reactor Netty 底层主要负责限制响应体内相邻数据之间的空闲间隔。对于 SSE 这种响应头很快就返回、body 持续流动的场景它直接决定了“服务端半天不说话”会不会被你当成故障。配置的超时越长你越要保证重试逻辑里有足够快的失败检测否则一段死连接会挂很久才发现。2.3 缓冲区与最大内存限制另一个容易踩的是maxInMemorySize。WebClient 默认对单条数据缓冲的字节数有限制默认值是 256KB。SSE 接口一旦出现特别大的单条 data——比如大模型某个流式片段带了一段超长 JSON或者日志接口一次推了 1MB 文本解码器会直接抛DataBufferLimitException表现就是你这边数据流突然中断排查半天发现根本不是业务问题。解决方式是在ExchangeStrategies里调大上限ExchangeStrategies strategies ExchangeStrategies.builder() .codecs(configurer - configurer.defaultCodecs().maxInMemorySize(4 * 1024 * 1024)) .build(); WebClient webClient WebClient.builder() .baseUrl(http://ai.example.com) .clientConnector(connector) .exchangeStrategies(strategies) .build();这里的限制是针对单个 DataBuffer 的不是整个流的累计上限。4MB 对绝大多数文本事件都足够如果业务里推送的是二进制大块内容甚至要分段读取那就得考虑自定义解码器了。从经验上说遇到“不定期中断 日志里没有业务异常”的问题时优先检查这个参数因为它抛出的异常很容易被下游的兜底逻辑吞掉。2.4 底层 ClientHttpConnector 的选型WebClient 底层用什么 Connector不是随便选的。默认情况下类路径里存在 Reactor Netty 时就用 Reactor Netty这个依赖也正因为spring-boot-starter-webflux被引入所以多数项目默认就是它。我个人的建议是不要为了省依赖去换 JDK HttpClient 作为底层来消费 SSE。JDK 自带的 HTTP 客户端对无限流式响应的处理细节比较苛刻我在一个项目里遇到过响应头已经返回、body 却迟迟不消费的现象换成 Reactor Netty 之后立刻正常。Reactor Netty 的流式处理和背压支持都更成熟长连接场景下稳定得多。如果你发现“SSE 数据迟迟不来”除了排查服务端也要检查是不是 Connector 选型发生了变化。3. 核心消费代码三种姿势与数据清洗3.1 最直接的姿势bodyToFlux(String.class) 收 data当服务端发的 SSE 事件比较简单你只关心 data 字段最省事的写法就是直接收字符串FluxString dataFlux webClient.get() .uri(/api/stream) .accept(MediaType.TEXT_EVENT_STREAM) .retrieve() .bodyToFlux(String.class); dataFlux.subscribe(data - log.info(收到数据: {}, data));这里必须提醒一个新手最容易懵的点Flux 是“冷”的你写了bodyToFlux却不调用subscribe请求根本不会发生代码会静默地“不工作”。很多从同步编程思维转过来的人在这个地方卡了很久还以为是接口没通。另一个坑是在验证阶段有人会顺手写.blockLast()去等“最后一个元素”。SSE 是无限流这么一写程序就永远不退出数据量大时还会把内存顶爆。想验证数据是否正确用subscribe打日志最安全不要为了省事去 block。WebClient 拿到text/event-stream响应后会自动用内部 SSE 解码器逐条拆事件每个事件的 data 部分作为一条元素发给下游。如果服务端在一个事件里发了多行 data它会按协议规范合并用换行符连接。很多刚接触的人以为一个事件只有一行 data拼 JSON 时发现内容断了其实是被协议行为拆开了。3.2 进阶姿势ServerSentEvent 拿全字段如果需要拿到事件 id、event 类型、注释等元信息甚至要做断点续传就不要直接转 String而是转成ServerSentEvent泛型对象ParameterizedTypeReferenceServerSentEventString sseType new ParameterizedTypeReferenceServerSentEventString() {}; webClient.get() .uri(/api/stream) .accept(MediaType.TEXT_EVENT_STREAM) .retrieve() .bodyToFlux(sseType) .subscribe(event - { String id event.id(); String eventName event.event(); String data event.data(); String comment event.comment(); log.info(id{}, event{}, data{}, id, eventName, data); });这里有个容易被忽略的技术细节必须用ParameterizedTypeReference包裹泛型而不是直接传ServerSentEvent.class。这样做是为了让 Spring 知道 data 字段是 String 类型否则拿到的会是 Object后面强转时容易埋雷。拿到这些字段后实际用途很清晰。id用于续传前端 EventSource 断开重连时会自动把最后收到的 id 放进Last-Event-ID请求头服务端据此决定从哪继续推event类似消息主题下游可以按事件类型做 filtercomment是注释行通常只用来做心跳不承载业务数据处理时要么忽略要么打进 debug 日志。3.3 响应式流水线过滤、转换、限速与限量拿到 Flux 之后数据清洗的爽点就在于可以用操作符搭一条流水线。比如上游除了业务数据还会发 ping 心跳事件你可以在转发前先过滤掉webClient.get() .uri(/api/stream) .accept(MediaType.TEXT_EVENT_STREAM) .retrieve() .bodyToFlux(sseType) .filter(ev - ev.comment() null) .map(ev - MyEvent.from(ev.data())) .doOnNext(event - metrics.record(sse_event, 1)) .subscribe(...)控制消费速率也在这里做。下游处理不过来时可以在链路上加limitRate(50)让上游维持大约 50 条一批的拉取节奏背压因此真正生效想只消费前 N 条就take(100)想跑一个固定时间窗口就takeUntilOther(Mono.delay(Duration.ofMinutes(5)))。这些操作符组合起来相当于把“流式连接的生命周期”纳入了代码控制换在同步模型里要自己写线程、写状态标志在响应式模型里就是几行声明。4. 转发与续传把上游 SSE 流安全地交给前端4.1 Spring MVC 场景用 SseEmitter 桥接如果你的主项目是 Spring MVC而前端是浏览器页面最常用的是SseEmitter。它是 Servlet 异步请求模型里的 SSE 输出器。思路是后端接口先创建一个 SseEmitter 返回给前端然后在这个接口背后订阅上游 WebClient 的 Flux每收到一条数据就写进 emitter。这里有两个关键点。第一前端连接断开后要主动取消上游订阅否则销毁的只是 emitter上游流还在后台跑着白白消耗资源。第二SseEmitter 的默认 async timeout 很短如果前端只是挂着页面看流式数据不设置超时的话连接会被服务端隔一阵掐一次。一个可用的版本大概长这样GetMapping(/proxy/live) public SseEmitter proxyLive() { SseEmitter emitter new SseEmitter(0L); Disposable disposable sseWebClient.get() .uri(/upstream/live) .accept(MediaType.TEXT_EVENT_STREAM) .retrieve() .bodyToFlux(String.class) .subscribe( data - safeSend(emitter, data), error - { log.error(上游异常, error); emitter.completeWithError(error); }, emitter::complete ); emitter.onTimeout(() - { log.warn(前端连接超时); disposable.dispose(); }); emitter.onError(error - { log.warn(前端连接异常, error); disposable.dispose(); }); return emitter; }safeSend里要捕获IOException。前端断开的瞬间后端可能还会继续写数据那一次 send 是会抛异常的。我真正踩过这个坑时才明白如果不捕获 IOException日志看起来会有一些杂散异常但更隐蔽的是大量连接在空转因为没有地方去终止上游订阅。捕获之后不要犹豫直接取消订阅并完成 emitter。private void safeSend(SseEmitter emitter, String data) { try { emitter.send(SseEmitter.event() .name(message) .data(data, MediaType.TEXT_PLAIN)); } catch (IOException e) { throw new RuntimeException(前端连接已断开, e); } }注意 send 时最好显式指定 MediaType否则默认行为是让框架根据数据内容自动选择转换器某些场景下前端 EventSource 会解析不到内容。4.2 WebFlux 场景直接返回 Flux 更省心如果项目本身就是 WebFlux或者你有新建响应式模块的空间事情会简单很多。Controller 直接返回FluxServerSentEventSpring 会把框架内部的背压机制和取消语义都处理好连 SseEmitter 都不用碰GetMapping(value /api/live, produces MediaType.TEXT_EVENT_STREAM_VALUE) public FluxServerSentEventString live() { return fluxFromUpstream() .map(data - ServerSentEvent.builder(data) .event(message) .build()); }直接返回 Flux 时框架会按下游消费节奏反压到上游前端断开也能顺着响应式链自动取消订阅。代码量最少错误处理也更干净。唯一要注意的是 filter 和 map 里的异常要有兜底否则整体链路异常会导致连接终止前端 EventSource 会立刻自动重连造成一种“疯狂重连”的错觉往往比断线本身更吓人。4.3 断线重试与 Last-Event-ID 续传SSE 协议本身有断线重连能力但 WebClient 作为客户端时重连逻辑需要自己写。我的建议是把事件 id 记下来失败后带着Last-Event-ID头重新请求。服务端如果支持续传就能从断点继续推而不是每次都从第一条重复消费。AtomicReferenceString lastEventId new AtomicReference(); FluxServerSentEventString resumable sseWebClient.get() .uri(/api/stream) .header(Last-Event-ID, lastEventId.get()) .accept(MediaType.TEXT_EVENT_STREAM) .retrieve() .bodyToFlux(sseType) .doOnNext(event - lastEventId.set(event.id())) .doOnError(err - log.error(SSE 连接异常, err)) .retryWhen(Retry.backoff(5, Duration.ofSeconds(2)) .maxBackoff(Duration.ofSeconds(30)) .jitter(0.5)); resumable.map(ServerSentEvent::data).subscribe(...);这里有两个细节。第一header(Last-Event-ID, lastEventId.get())在请求发出时取的是当时的 id必须用AtomicReference这类可变容器保存最新 id普通 String 变量在闭包里是失效的。第二retryWhen的每次重试都会产生一个新的订阅、重新发起请求网络抖动时如果 backoff 太短会造成连接风暴。jitter(0.5)在这里不是装饰它能打散多客户端 reconnect 的并发冲击对接多个渠道时效果尤其明显。5. 高频踩坑排查与建议5.1 常见问题对照表我把这个方案里最容易翻车的现象和解决办法汇总成一张表后面再展开讲两个典型。现象常见原因处理方式调用后没有任何输出没订阅 Flux或上游 Content-Type 不是 text/event-stream检查 subscribe 与响应头用 accept 显式声明单条数据很大时报 DataBufferLimitException默认 maxInMemorySize 偏小在 ExchangeStrategies 调大缓冲上限中文乱码上游字符集声明缺失或错误要求上游返回 charsetutf-8或自定义解码逻辑SseEmitter 每隔 30 秒左右断开Servlet 异步请求超时new SseEmitter(0L) 并合理设置超时前端断线后日志还在刷数据没在 onTimeout/onError 中取消订阅保存 Disposable连接终结时调 dispose进程内存缓慢上涨无限流未做 take/limitRate积压到内存按业务窗口限量、限速关注背压长时间无数据后连接被掐断responseTimeout 设置过短调大 responseTimeout或依赖心跳兜底重连后重复消费大量数据没实现 Last-Event-ID 续传记录最后 id重连请求带该 Header这张表几乎覆盖了我多年 SSE 集成里遇到的所有线上问题。你会发现大部分都不是新技术难题而是对流式连接生命周期的理解不足、配置随意最终表现为玄学问题。其实每个现象都能回到 Http 协议、Flux 订阅模型和 Servlet 异步请求这几个基础概念上。5.2 一个典型排查案例日志一直不输出有一次我接一个第三方实时日志接口Postman 里访问完全正常写进 Spring Boot 以后一条数据都没有。排查了几个小时最后定位到两个原因。第一接口返回的 Content-Type 是text/plain虽然 body 里是标准的 SSE 格式但 WebClient 看到text/plain不会启用 SSE 解码器数据被当成普通字符串整体缓冲当然不会按事件流输出。解决办法是在请求里显式accept(MediaType.TEXT_EVENT_STREAM)同时对返回的 content-type 做兼容判断避免解码器报错。第二读超时设置过短连接刚建立就被 client 自己掐断。所以我后来有个习惯对接任何第三方 SSE 之前先 curl 一下看响应头比在代码里埋点找问题快得多。头信息不对后面再多代码都是沙滩上的楼。5.3 我在项目里养成的几条实践习惯最后分享几个我从实战里沉淀下来的习惯。第一对接外部 SSE 前先确认对方有没有心跳。没有心跳客户端超时就要放宽前端也要有兜底重连机制。第二SSE 流的管道里不要做耗时操作。filter、map 里如果写数据库、调外部接口背压会帮你把延迟一路堆积到上游最终表现为服务端推送越来越慢。要处理数据就先把它分流到异步消息队列让消费链保持轻量。第三给上游服务建一层不变量监控断线次数、单条数据最大字节数、每秒事件数、重试间隔都落到日志和指标里。长连接的问题往往要故障半小时后才能从用户反馈中得知监控是帮你提前发现问题的唯一手段。我现在做代码评审时只要看到 WebClient 消费 SSE 的地方都会确认三件事订阅后的 Disposable 存没存、timeout 有没有兜底、重连有没有带续传 id。把这三个点盯住后面能少加很多班。SSE 本身不是什么高深技术但长连接这种形态天生会把配置疏漏和资源管理问题放大能提前改掉的就别等到生产环境给你答案。