SSE：Server-Sent Events
SSE 是服务器向客户端推送实时事件的标准协议。基于 HTTP/1.1，使用 text/event-stream MIME 类型，每个事件格式为：
event: tick
data: {"seq": 1, "time": 1700000000000}
id: 42
event: price
data: {"symbol": "BTC", "price": 42000}
:heartbeat
WebFlux 的 SSE 支持极其简洁：
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;
import java.time.Duration;
@RestController
public class SseController {
@GetMapping(value = "/sse/ticks", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<String> ticks() {
return Flux.interval(Duration.ofSeconds(1))
// Spring 会把每个元素当作 SSE 的 data 字段并自动补 "data:"，
// 这里不要再手写 data: 和空行
.map(i -> "{\"seq\":" + i + ",\"time\":" + System.currentTimeMillis() + "}");
}
}
浏览器用 EventSource 消费：
const es = new EventSource('/sse/ticks');
es.onmessage = (e) => {
const data = JSON.parse(e.data);
console.log('seq:', data.seq, 'time:', data.time);
};
带事件类型的 SSE
上面的例子返回 Flux<String>，Spring 会把每个字符串当作 data 字段，自动补 data: 前缀和空行；如果字符串里再手写 data:，客户端会收到 data:data:...。要控制 id、event、retry 这些字段，才需要换成 ServerSentEvent：
import org.springframework.http.codec.ServerSentEvent;
@GetMapping(value = "/sse/events", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<ServerSentEvent<String>> events() {
return Flux.interval(Duration.ofSeconds(1))
.map(i -> ServerSentEvent.<String>builder()
.id(String.valueOf(i))
.event("tick")
.data("seq=" + i)
.build());
}
ServerSentEvent 会自动生成格式化的 SSE 帧：
id: 0
event: tick
data: seq=0
id: 1
event: tick
data: seq=1
Flux 响应与背压
Flux 返回值的背压如何工作？Spring WebFlux 内部把 Flux 订阅到 HTTP 输出流，订阅的 request(n) 与 Netty 的 isWritable() 联动：
@GetMapping(value = "/sse/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<String> stream() {
return Flux.interval(Duration.ofMillis(100))
.map(i -> "event-" + i);
}
执行链：
Flux.interval(100ms) ← 数据源每 100ms 生成一个 tick
↓ onBackpressure... ← WebFlux 自动背压
HttpMessageWriter ← Spring 的 SSE writer
↓ request(n) ← 根据 TCP 可写状态动态调整
Reactor Netty HttpServerResponse
↓ writeWith() ← Netty ByteBuf 缓冲区
TCP send buffer ← 滑动窗口
↓
客户端 EventSource ← onmessage 回调消费
如果客户端处理慢（如 JS 回调中有繁重计算），EventSource 的内部缓冲会积压，TCP 接收窗口缩小，Netty 写缓冲变得不可写，WebFlux 的 writer 随之减少向上游的 request(n)。但要注意：Flux.interval 是定时源，不会因为下游请求变少而降低发射频率；下游没有请求时，它静默丢弃当前 tick（debug 日志里能看到 Dropping tick），不会把元素留到下一次。需要保留每个 tick 时，应在 interval 之后接 onBackpressureBuffer()（并设上限），或者改成由业务事件驱动的 Sinks.Many。真正「按需生产」的源（如数据库游标、文件读取）才会自然跟随 request(n) 降速。
chunked Transfer-Encoding
SSE 是 text/event-stream 的持续响应，不是 chunked 的「特殊形式」：HTTP/1.1 下它通常由 Transfer-Encoding: chunked 承载，HTTP/2 下则直接走流式帧，没有 chunked 这一层。对于普通 HTTP 响应，也可以返回 Flux<DataBuffer> 做流式传输：
@GetMapping(value = "/download/large", produces = MediaType.APPLICATION_OCTET_STREAM_VALUE)
public Flux<DataBuffer> downloadLarge() {
return DataBufferUtils.read(
Path.of("/tmp/large-file.zip"),
new DefaultDataBufferFactory(),
8192 // 8KB buffer
);
}
DataBufferUtils.read() 返回 Flux<DataBuffer>。HTTP/1.1 下响应没有 Content-Length，Spring 会用 chunked 分块输出；HTTP/2 下由协议帧承担分块。客户端逐块接收，无需等待整个文件加载。注意这是「分块读取 + 分块发送」，不是零拷贝：文件内容仍要经过 DataBuffer 和 Netty 的写路径。
自定义 Content-Type 协商
WebFlux 根据 Accept 请求头和 produces 属性协商响应格式：
@GetMapping(value = "/report", produces = {
MediaType.APPLICATION_JSON_VALUE,
"text/csv"
})
public Mono<Report> report(ServerWebExchange exchange) {
List<MediaType> accepted = exchange.getRequest().getHeaders().getAccept();
if (accepted.stream().anyMatch(mt -> mt.isCompatibleWith(MediaType.valueOf("text/csv")))) {
return generateCsv();
}
return generateJson();
}
更正式的做法是使用 ContentNegotiationConfigurer：
@Configuration
public class WebConfig implements WebFluxConfigurer {
@Override
public void configureContentNegotiation(ContentNegotiationConfigurer configurer) {
configurer
.favorParameter(true) // ?format=json 优先级高于 Accept 头
.parameterName("format")
.defaultContentType(MediaType.APPLICATION_JSON);
}
}
响应式错误与 SSE
SSE 连接的错误处理比单次 HTTP 请求更复杂。如果 Flux 发出 onError，服务端响应以错误终止，连接关闭；浏览器会触发传输层 error 事件，并按规范默认在约 3 秒后自动重连（服务端可用 retry: 字段覆盖间隔）。把业务错误封装为事件只是让客户端能读到原因，并不会阻止自动重连：
@GetMapping(value = "/sse/safe", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<ServerSentEvent<Object>> safeStream() {
return riskyDataSource()
.map(data -> ServerSentEvent.<Object>builder()
.event("data")
.data(data)
.build())
.onErrorResume(err -> Flux.just(
ServerSentEvent.<Object>builder()
.event("error")
.data(Map.of("message", err.getMessage()))
.build()
));
}
客户端收到 event: error 后可以决定是否重连：
es.addEventListener('error', (e) => {
if (typeof e.data === 'string') {
// 服务端发送的 event: error，带业务错误数据
console.log('Stream error:', JSON.parse(e.data));
// 业务错误不需要重连时显式关闭；否则浏览器会自动重连
es.close();
} else {
// 传输层错误（断网、服务端关闭）没有 data，浏览器会自动重连
console.warn('SSE transport error; browser will retry');
}
});
注意 addEventListener('error') 同时会收到传输层错误和服务端自定义的 event: error，两者的区别是后者带 e.data。不想重连时，客户端调用 es.close()，或服务端在重连请求上返回 204 No Content。
SSE vs WebSocket 选择
| 维度 | SSE | WebSocket |
|---|---|---|
| 方向 | 单向：服务器 → 客户端 | 双向 |
| 协议 | HTTP/1.1，自动重连 | 独立协议，需手动重连 |
| 穿透代理 | 好（纯 HTTP） | 差（某些 proxy 不支持 upgrade） |
| 消息格式 | 文本（JSON） | 文本或二进制 |
| 广播 | 每个客户端独立连接 | 可共享连接做广播 |
eventd v2 的实时推送同时使用两者：
- SSE：服务端事件通知（价格变动、订单状态更新）——单向、自动重连
- WebSocket：交互式场景（聊天、协同编辑）——双向、低延迟
测试 SSE 端点
用 curl 测试：
curl -N http://localhost:8080/sse/ticks
# data:{"seq":0,"time":1700000000000}
#
# data:{"seq":1,"time":1700000000100}
-N（--no-buffer）关闭 curl 的输出缓冲，保证实时看到事件。
用 WebTestClient 测试（第 6.3 节）：
webTestClient.get().uri("/sse/ticks")
.accept(MediaType.TEXT_EVENT_STREAM)
.exchange()
.expectStatus().isOk()
.expectHeader().contentTypeCompatibleWith(MediaType.TEXT_EVENT_STREAM)
.returnResult(String.class)
.getResponseBody()
.take(3)
.collectList()
.block();
小结
| 能力 | API | 注意 |
|---|---|---|
| 基本 SSE | @GetMapping(produces = TEXT_EVENT_STREAM) + Flux | Spring 自动补 data:，不要手写 |
| 结构化 SSE | Flux<ServerSentEvent<T>> | 自动格式化 id/event/data/retry |
| 背压 | WebFlux 内部自动 | 客户端慢 → request(n) 减小；定时源不会自动降速 |
| chunked | Flux<DataBuffer> | 大文件流式分块，不是零拷贝 |
| 格式协商 | ContentNegotiationConfigurer | ?format=json 或 Accept 头 |
| 错误处理 | onErrorResume 转为 error 事件 | 避免 EventSource 自动关闭 |
下一节 5.3 错误契约 ：全局异常处理器、统一 JSON 错误体、错误码体系、与 HTTP 状态码的映射。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。