Spring Boot 4.0 SSE 与 WebSocket 实时消息推送

作者：苏三2026-11-06Spring Boot 4.0SSEWebSocket实时消息Spring MVC

前言

查询接口遵循“客户端请求一次、服务端返回一次”。订单状态变化后，如果页面想知道最新金额或状态，只能不断轮询。

轮询实现简单，但请求多、实时性差。需要持续推送时，可以从两种传输中选：

SSE：基于 HTTP，服务端单向持续发送事件。
WebSocket：建立双向长连接，双方都可以主动发送消息。

本章会让订单创建和更新事件同时进入两条通道，并验证消息只会发给对应客户。

配套源码：dvsusan/spring-boot4-demo。

最终结果

SSE 订阅地址：

GET /api/v1/events/orders?customerId=C1001

WebSocket 地址：

ws://localhost:8080/ws/orders?customerId=C1001

订阅 C1001 后创建该客户的订单，SSE 收到：

event:order-created
data:{"eventId":"d03a0f59-99f5-470c-ab07-d921d1ebf85f","type":"order-created","orderNo":"SOD7D5107CB2AA","customerId":"C1001","status":"CREATED","occurredAt":"2026-11-06T05:03:32.866086Z"}

WebSocket 收到同样的 JSON 事件。

业务发布一次事件图

图 1：业务只发布一次事件，SSE 和 WebSocket 负责不同客户端传输。

第 1 步：先选择传输方式

对比项	SSE	WebSocket
通信方向	服务端到客户端	双向
协议基础	HTTP `text/event-stream`	WebSocket 升级握手
自动重连	浏览器 `EventSource` 原生支持	需要客户端实现
二进制	不适合	支持
代理兼容	通常更好	长连接配置更多
典型场景	通知、日志、状态流	聊天、协同编辑、交互控制

订单状态通知只需要服务端推送，SSE 已经足够。WebSocket 更适合用户还要持续发送指令的场景。

第 2 步：添加 WebSocket Starter

demo-web/pom.xml：

<dependency>
    <!-- 提供 Spring WebSocket 服务端与握手基础设施。 -->
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-websocket</artifactId>
</dependency>

SSE 由 Spring MVC 的 SseEmitter 支持，不需要额外依赖。

第 3 步：定义统一订单事件

创建：

demo-web/src/main/java/com/susan/training/event/OrderEvent.java

/**
 * 订单实时事件，SSE 和 WebSocket 共用同一结构。
 */
public record OrderEvent(
        String eventId,
        String type,
        String orderNo,
        String customerId,
        String status,
        Instant occurredAt
) {
}

事件包含 eventId，客户端可以用于去重。type 区分创建与更新，occurredAt 帮助乱序或延迟排查。

第 4 步：建立进程内事件分发器

创建：

demo-web/src/main/java/com/susan/training/event/OrderEventBroker.java

核心结构：

@Component
public class OrderEventBroker {
    private final ObjectMapper objectMapper;

    // 每个 SSE 连接保存订阅的客户编号。
    private final Map<String, SseSubscription> sseSubscriptions =
            new ConcurrentHashMap<>();

    // 每个 WebSocket 会话也保存相同维度的订阅信息。
    private final Map<String, WebSocketSubscription> webSocketSubscriptions =
            new ConcurrentHashMap<>();

    public OrderEventBroker(ObjectMapper objectMapper) {
        this.objectMapper = objectMapper;
    }
}

创建 SSE 连接：

public SseEmitter subscribeSse(String customerId) {
    String subscriptionId = UUID.randomUUID().toString();
    SseEmitter emitter = new SseEmitter(30_000L);
    sseSubscriptions.put(
            subscriptionId,
            new SseSubscription(customerId, emitter)
    );

    // 连接完成、超时或报错时都必须移除引用。
    emitter.onCompletion(() -> sseSubscriptions.remove(subscriptionId));
    emitter.onTimeout(() -> sseSubscriptions.remove(subscriptionId));
    emitter.onError(error -> sseSubscriptions.remove(subscriptionId));

    try {
        // 首条 connected 事件用于确认订阅已经生效。
        emitter.send(SseEmitter.event()
                .name("connected")
                .data(Map.of(
                        "customerId", customerId,
                        "transport", "sse"
                )));
    } catch (IOException exception) {
        sseSubscriptions.remove(subscriptionId);
        emitter.completeWithError(exception);
    }
    return emitter;
}

发布事件：

// 一次业务变更生成一个事件，再分别投递到两种连接。
public void publish(String eventType, OrderView order) {
    OrderEvent event = new OrderEvent(
            UUID.randomUUID().toString(),
            eventType,
            order.orderNo(),
            order.customerId(),
            order.status().name(),
            Instant.now()
    );
    sendToSse(event);
    sendToWebSocket(event);
}

发送时按 customerId 过滤：

private void sendToSse(OrderEvent event) {
    sseSubscriptions.forEach((id, subscription) -> {
        if (!subscription.customerId().equals(event.customerId())) {
            return;
        }
        try {
            subscription.emitter().send(SseEmitter.event()
                    .id(event.eventId())
                    .name(event.type())
                    .data(event));
        } catch (Exception exception) {
            // 客户端可能已经断开，继续广播时先清理失效连接。
            sseSubscriptions.remove(id);
        }
    });
}

发送到 WebSocket：

private void sendToWebSocket(OrderEvent event) {
    webSocketSubscriptions.forEach((id, subscription) -> {
        if (!subscription.customerId().equals(event.customerId())
                || !subscription.session().isOpen()) {
            return;
        }
        try {
            String payload = objectMapper.writeValueAsString(event);
            subscription.session().sendMessage(new TextMessage(payload));
        } catch (Exception exception) {
            // 序列化或网络发送失败后，不保留失效会话。
            webSocketSubscriptions.remove(id);
        }
    });
}

public void unregisterWebSocket(String sessionId) {
    webSocketSubscriptions.remove(sessionId);
}

@PreDestroy
public void closeAll() {
    // 正常停机时主动完成 SSE，浏览器可以尽快进入重连流程。
    sseSubscriptions.values()
            .forEach(subscription -> subscription.emitter().complete());
    sseSubscriptions.clear();
    webSocketSubscriptions.clear();
}

private record SseSubscription(
        String customerId,
        SseEmitter emitter
) {
}

private record WebSocketSubscription(
        String customerId,
        WebSocketSession session
) {
}

生产环境还要考虑发送队列、慢客户端和背压。本章使用 SseEmitter 做最小可运行模型，不把它当成完整消息平台。

第 5 步：提供 SSE 接口

创建：

demo-web/src/main/java/com/susan/training/web/stream/OrderEventController.java

@RestController
@RequestMapping("/api/v1/events/orders")
public class OrderEventController {
    private final OrderEventBroker eventBroker;

    public OrderEventController(OrderEventBroker eventBroker) {
        this.eventBroker = eventBroker;
    }

    @GetMapping(produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public SseEmitter subscribe(
            @RequestParam @NotBlank String customerId
    ) {
        // 方法立即返回 emitter，HTTP 连接在后台保持打开。
        return eventBroker.subscribeSse(customerId);
    }
}

produces = text/event-stream 是 SSE 协议的关键。返回类型必须是 SseEmitter，不能先包成普通 ApiResponse。

第 6 步：处理 WebSocket 连接

创建：

demo-web/src/main/java/com/susan/training/web/stream/OrderWebSocketHandler.java

// 每个 WebSocket 会话在连接建立时进入事件分发器。
@Component
public class OrderWebSocketHandler extends TextWebSocketHandler {
    private final OrderEventBroker eventBroker;
    private final ObjectMapper objectMapper;

    public OrderWebSocketHandler(
            OrderEventBroker eventBroker,
            ObjectMapper objectMapper
    ) {
        this.eventBroker = eventBroker;
        this.objectMapper = objectMapper;
    }

    @Override
    public void afterConnectionEstablished(
            WebSocketSession session
    ) throws Exception {
        String customerId = UriComponentsBuilder.fromUri(session.getUri())
                .build()
                .getQueryParams()
                .getFirst("customerId");
        if (customerId == null || customerId.isBlank()) {
            session.close(
                    CloseStatus.BAD_DATA.withReason("customerId is required")
            );
            return;
        }
        eventBroker.registerWebSocket(customerId, session);
        session.sendMessage(new TextMessage(
                objectMapper.writeValueAsString(Map.of(
                        "type", "connected",
                        "customerId", customerId,
                        "transport", "websocket"
                ))
        ));
    }
}

断开时清理：

@Override
public void afterConnectionClosed(
        WebSocketSession session,
        CloseStatus status
) {
    // 防止已关闭会话一直留在 Map 中。
    eventBroker.unregisterWebSocket(session.getId());
}

第 7 步：注册 WebSocket 地址

创建：

demo-web/src/main/java/com/susan/training/config/OrderWebSocketConfiguration.java

@Configuration
@EnableWebSocket
public class OrderWebSocketConfiguration
        implements WebSocketConfigurer {
    private final OrderWebSocketHandler orderWebSocketHandler;

    public OrderWebSocketConfiguration(
            OrderWebSocketHandler orderWebSocketHandler
    ) {
        this.orderWebSocketHandler = orderWebSocketHandler;
    }

    @Override
    public void registerWebSocketHandlers(
            WebSocketHandlerRegistry registry
    ) {
        // 示例允许本地页面；生产环境必须限制前端域名。
        registry.addHandler(orderWebSocketHandler, "/ws/orders")
                .setAllowedOriginPatterns("*");
    }
}

setAllowedOriginPatterns("*") 只适合教学。生产环境应列出明确域名，并结合认证、授权和连接数限制。

第 8 步：在订单变更后发布事件

在 OrderController 中注入事件分发器：

// Controller 在业务成功后发布实时事件。
private final OrderEventBroker eventBroker;

public OrderController(
        OrderService orderService,
        OrderEventBroker eventBroker
) {
    this.orderService = orderService;
    this.eventBroker = eventBroker;
}

创建成功后发布：

// 创建成功后再发布事件，不在失败请求中广播。
OrderView order = orderService.create(
        new CreateOrderCommand(request.customerId(), request.amount())
);
eventBroker.publish("order-created", order);
return ApiResponse.success(order);

更新成功后发布：

// 更新成功后广播最新订单快照。
OrderView updated = orderService.update(
        orderNo,
        request.customerId(),
        request.amount()
);
eventBroker.publish("order-updated", updated);
return ApiResponse.success(updated);

事件发布应放在业务结果成功之后。当前内存服务在 create/update 返回时已经完成变更；接入数据库后，要明确提交事务与发布事件的顺序。

第 9 步：验证 SSE

构建并启动：

mvn clean package -DskipTests
java -jar demo-web/target/demo-web-1.0.0-SNAPSHOT.jar

终端一订阅：

curl -N \
  "http://localhost:8080/api/v1/events/orders?customerId=C1001"

第一行事件：

event:connected
data:{"transport":"sse","customerId":"C1001"}

终端二创建订单：

curl -X POST "http://localhost:8080/api/v1/orders" \
  -H "Content-Type: application/json" \
  -d '{"customerId":"C1001","amount":99.00}'

终端一应在 30 秒连接窗口内收到 event:order-created。

用其他客户创建订单，例如 C2002，当前订阅不应收到该事件。

第 10 步：验证 WebSocket

当前机器可安装 ws 客户端后执行：

mkdir -p /tmp/order-ws
npm install --prefix /tmp/order-ws --no-save ws@8.18.3

Node.js 验证脚本：

const WebSocket = require('/tmp/order-ws/node_modules/ws');

// 查询参数决定订阅哪个客户的事件。
const socket = new WebSocket(
  'ws://localhost:8080/ws/orders?customerId=C1001'
);

socket.on('message', async (raw) => {
  const message = JSON.parse(raw.toString());
  console.log(message);

  // 收到 connected 后再触发一笔订单，确保订阅已建立。
  if (message.type === 'connected') {
    await fetch('http://localhost:8080/api/v1/orders', {
      method: 'POST',
      headers: {'Content-Type': 'application/json'},
      body: JSON.stringify({customerId: 'C1001', amount: 128.00})
    });
  }

  if (message.type === 'order-created') {
    socket.close();
  }
});

预期依次收到：

{ transport: 'websocket', type: 'connected', customerId: 'C1001' }
{ eventId: '...', type: 'order-created', orderNo: 'SO...', ... }

本项目已经用相同方式完成真实 WebSocket 连接和订单事件验证。

常见问题排查

现象	常见原因	处理方式
SSE 一直只有 `connected`	订单客户与订阅客户不同	检查两个请求的 `customerId`
网关迟迟不返回事件	中间层缓冲了文本流	关闭该路由的响应缓冲
事件发不出且调用返回 500	断开连接仍在订阅集合中	发送失败时移除连接
WebSocket 握手 400	`customerId` 缺失	在连接 URL 中携带查询参数
WebSocket 突然断开	代理空闲超时	增加心跳并按指数退避重连
多实例只收到一部分消息	事件只在当前 JVM 内广播	接入 Redis Pub/Sub、RocketMQ 等广播层

本章小结

SSE 和 WebSocket 解决不同问题：

单向状态流优先 SSE。
双向交互使用 WebSocket。
两者都可以复用同一个业务事件模型。

本章实现是单实例教学版。生产系统还要补齐认证、连接数限制、心跳、重连、背压、消息持久化和多实例广播。

练习题

为订单事件增加心跳：

每 15 秒向 SSE 客户端发送 event: heartbeat。
WebSocket 每 15 秒发送一次 ping。
连续两次发送失败时移除连接。
客户端断线后按 1、2、4、8 秒退避重连。

上一节：RestClient 与 HTTP Interface
返回课程目录