Spring Boot 响应式 SSE 实时推送，单机吞吐量10万+

在高并发场景下，Spring Boot 结合 响应式编程（WebFlux） + SSE（Server-Sent Events） 是实现实时推送的最优方案之一 ——SSE 基于 HTTP 长连接，仅服务器向客户端单向推送，开销远低于 WebSocket（无需双工通信），而 WebFlux 的非阻塞响应式模型能支撑海量并发连接（单机轻松突破 10 万 + 吞吐量）。

一、核心原理：为什么能支撑 10 万 + 并发？

1. SSE vs WebSocket：选对场景是关键

特性	SSE（Server-Sent Events）	WebSocket
通信方向	单向（服务器→客户端）	双向（全双工）
协议基础	基于 HTTP（复用端口、兼容代理）	独立 TCP 协议（需额外端口）
连接开销	低（长连接复用，无握手开销）	中（需 TCP 握手 + WebSocket 握手）
并发支撑	高（非阻塞模型下连接数无上限）	中（双工通信占用更多资源）
适用场景	实时通知、行情推送、日志流	聊天、实时协作（需双向交互）

结论：如果仅需服务器向客户端单向推送（如订单通知、实时数据看板），SSE 是更轻量、高并发的选择。

2. 响应式 WebFlux：突破传统 MVC 瓶颈

传统 Spring MVC 基于 Servlet 模型（阻塞 IO），每个请求对应一个线程，线程数上限（默认 200）直接限制并发连接数（无法支撑 10 万 +）。

而 Spring WebFlux 基于 Reactor 响应式框架（非阻塞 IO），核心是：

用少量 EventLoop 线程（默认 CPU 核心数 ×2）处理海量连接（IO 等待时释放线程，不阻塞）；
基于 Flux/Mono 数据流模型，支持背压（Backpressure），避免客户端处理不及导致的服务器过载。

两者核心差异：

模型	线程模型	并发支撑	适用场景
Spring MVC	每请求一线程（阻塞）	低（线程数上限）	同步短请求
Spring WebFlux	事件循环（非阻塞）	极高（百万级连接）	异步长连接 / 流处理

二、实战实现：Spring Boot + WebFlux + SSE

1. 环境依赖（Maven）

核心依赖
spring-boot-starter-webflux（响应式 Web 框架），无需额外 SSE 依赖（WebFlux 原生支持）：

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>3.2.0</version> <!-- 推荐 3.x（支持 Reactor 3.6+） -->
</parent>

<dependencies>
    <!-- 响应式 Web 核心依赖 -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-webflux</artifactId>
    </dependency>
    <!-- 工具类（可选） -->
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <optional>true</optional>
    </dependency>
</dependencies>

2. 核心实现：SSE 推送服务

需解决 3 个核心问题：

客户端连接管理（建立 / 销毁连接，关联用户 ID）；
实时数据流生成（模拟业务数据推送）；
连接安全与心跳（避免空闲连接被断开）。

（1）SSE 连接管理器（核心）

用 线程安全的 Map 存储用户与 SSE 连接的映射，支持连接注册、移除、批量推送：

import lombok.extern.slf4j.Slf4j;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.stereotype.Component;
import reactor.core.publisher.FluxSink;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;

/**
 * SSE 连接管理器：管理用户与 SSE 连接的映射
 */
@Component
@Slf4j
public class SseConnectionManager {

    /**
     * 存储用户ID -> SSE 数据流发射器（FluxSink）
     * ConcurrentHashMap 保证线程安全，支持高并发读写
     */
    private final Map<String, FluxSink<ServerSentEvent<String>>> userSinks = new ConcurrentHashMap<>();

    /**
     * 注册用户连接
     * @param userId 用户唯一标识
     * @param sink SSE 数据流发射器（由 WebFlux 生成）
     */
    public void register(String userId, FluxSink<ServerSentEvent<String>> sink) {
        // 注册前先移除旧连接（避免重复连接）
        remove(userId);
        // 注册新连接，并添加断开回调（客户端断开时移除连接）
        userSinks.put(userId, sink);
        sink.onDispose(() -> {
            log.info("用户[{}] SSE 连接断开，移除连接", userId);
            userSinks.remove(userId);
        });
        log.info("用户[{}] SSE 连接注册成功，当前在线数：{}", userId, userSinks.size());
    }

    /**
     * 移除用户连接
     * @param userId 用户唯一标识
     */
    public void remove(String userId) {
        FluxSink<ServerSentEvent<String>> sink = userSinks.remove(userId);
        if (sink != null) {
            sink.dispose(); // 主动销毁数据流
            log.info("用户[{}] SSE 连接已移除", userId);
        }
    }

    /**
     * 向单个用户推送消息
     * @param userId 用户唯一标识
     * @param message 推送内容
     */
    public void pushToUser(String userId, String message) {
        FluxSink<ServerSentEvent<String>> sink = userSinks.get(userId);
        if (sink != null && !sink.isCancelled()) {
            try {
                // 构建 SSE 消息（id：消息唯一标识，data：消息内容，retry：重连间隔）
                ServerSentEvent<String> sse = ServerSentEvent.<String>builder()
                        .id(String.valueOf(System.currentTimeMillis())) // 消息ID（客户端可记录，重连时请求未接收的消息）
                        .data(message)
                        .retry(3000) // 客户端断开后，3秒后自动重连
                        .build();
                sink.next(sse); // 发送消息
            } catch (Exception e) {
                log.error("向用户[{}]推送消息失败", userId, e);
                remove(userId); // 推送失败，移除无效连接
            }
        }
    }

    /**
     * 广播消息（向所有在线用户推送）
     * @param message 广播内容
     */
    public void broadcast(String message) {
        if (userSinks.isEmpty()) return;
        // 遍历所有连接推送（CopyOnWriteArraySet 避免遍历中修改集合报错）
        new CopyOnWriteArraySet<>(userSinks.entrySet()).forEach(entry -> {
            String userId = entry.getKey();
            FluxSink<ServerSentEvent<String>> sink = entry.getValue();
            if (!sink.isCancelled()) {
                try {
                    ServerSentEvent<String> sse = ServerSentEvent.<String>builder()
                            .id(String.valueOf(System.currentTimeMillis()))
                            .data(message)
                            .retry(3000)
                            .build();
                    sink.next(sse);
                } catch (Exception e) {
                    log.error("向用户[{}]广播消息失败", userId, e);
                    remove(userId);
                }
            }
        });
        log.info("SSE 广播消息成功，推送用户数：{}", userSinks.size());
    }

    /**
     * 获取当前在线用户数
     */
    public int getOnlineCount() {
        return userSinks.size();
    }
}

（2）SSE 控制器（暴露接口）

提供 连接建立接口 和 测试推送接口，基于 WebFlux 的 Flux<ServerSentEvent> 响应数据流：

import lombok.RequiredArgsConstructor;
import org.springframework.http.MediaType;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Flux;

import java.time.Duration;

/**
 * SSE 控制器：暴露 HTTP 接口供客户端连接和测试推送
 */
@RestController
@RequestMapping("/sse")
@RequiredArgsConstructor
public class SseController {

    private final SseConnectionManager connectionManager;

    /**
     * 建立 SSE 连接（核心接口）
     * 1. MediaType.TEXT_EVENT_STREAM：SSE 标准响应类型
     * 2. Flux<ServerSentEvent<String>>：响应式数据流，持续向客户端推送消息
     */
    @GetMapping(value = "/connect/{userId}", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<String>> connect(@PathVariable String userId) {
        // 创建 Flux 数据流，当客户端订阅时触发回调（注册连接）
        return Flux.create(sink -> {
            // 注册用户连接
            connectionManager.register(userId, sink);
            // 发送连接成功的欢迎消息
            ServerSentEvent<String> welcomeSse = ServerSentEvent.<String>builder()
                    .id(String.valueOf(System.currentTimeMillis()))
                    .data("SSE 连接成功，用户ID：" + userId)
                    .build();
            sink.next(welcomeSse);
        })
        // 心跳机制：每30秒发送一次空消息，避免连接被防火墙/网关断开
        .mergeWith(Flux.interval(Duration.ofSeconds(30))
                .map(seq -> ServerSentEvent.<String>builder()
                        .id("heartbeat-" + seq)
                        .data("") // 空消息（客户端可忽略）
                        .build()))
        // 超时配置：30分钟无数据推送则断开连接（可选）
        .timeout(Duration.ofMinutes(30));
    }

    /**
     * 向单个用户推送消息（测试接口）
     */
    @PostMapping("/push/{userId}")
    public String pushToUser(@PathVariable String userId, @RequestParam String message) {
        connectionManager.pushToUser(userId, message);
        return "消息推送成功，当前在线数：" + connectionManager.getOnlineCount();
    }

    /**
     * 广播消息（测试接口）
     */
    @PostMapping("/broadcast")
    public String broadcast(@RequestParam String message) {
        connectionManager.broadcast(message);
        return "广播成功，当前在线数：" + connectionManager.getOnlineCount();
    }

    /**
     * 获取在线用户数
     */
    @GetMapping("/onlineCount")
    public int getOnlineCount() {
        return connectionManager.getOnlineCount();
    }
}

（3）客户端测试（HTML/JS）

SSE 客户端原生支持（无需额外依赖），通过 EventSource 建立连接并监听消息：

<!DOCTYPE html>
<html>
<head>
    <title>SSE 实时推送测试</title>
    <meta charset="UTF-8">
</head>
<body>
    <h1>SSE 消息接收区</h1>
    <div id="message"></div>

    <script>
        // 用户ID（模拟登录用户，实际应从登录态获取）
        const userId = "user-" + Math.random().toString(36).substr(2, 9);
        console.log("当前用户ID：", userId);

        // 建立 SSE 连接（注意：协议为 http/https，端口与服务端一致）
        const eventSource = new EventSource(`http://localhost:8080/sse/connect/${userId}`);

        // 连接成功回调
        eventSource.onopen = function (e) {
            console.log("SSE 连接建立成功", e);
            appendMessage("✅ 连接建立成功，用户ID：" + userId);
        };

        // 接收消息回调
        eventSource.onmessage = function (e) {
            console.log("收到消息：", e.data);
            if (e.data) { // 忽略心跳空消息
                appendMessage(" " + new Date().toLocaleString() + "：" + e.data);
            }
        };

        // 连接错误回调（自动重连）
        eventSource.onerror = function (e) {
            console.error("SSE 连接错误", e);
            appendMessage("❌ 连接异常，正在重连...");
        };

        // 渲染消息到页面
        function appendMessage(content) {
            const div = document.createElement("div");
            div.style.margin = "8px 0";
            div.style.padding = "8px";
            div.style.borderLeft = "3px solid #2196F3";
            div.style.backgroundColor = "#f5f5f5";
            div.innerText = content;
            document.getElementById("message").appendChild(div);
        }
    </script>
</body>
</html>

3. 关键配置优化（支撑 10 万 + 并发）

默认配置无法支撑海量连接，需调整 JVM、Netty、连接超时等参数：

（1）application.yml 核心配置

spring:
  webflux:
    server:
      max-http-header-size: 8KB # 最大请求头大小（默认足够）
      compression:
        enabled: true # 开启响应压缩（减少数据传输量）
  reactor:
    netty:
      http:
        server:
          # Netty  acceptor 线程数（默认 CPU 核心数，无需修改）
          # 工作线程数（默认 CPU 核心数×2，无需修改）
          max-connections: 100000 # 最大并发连接数（关键：设置为 10 万）
          connection-timeout: 60s # 连接超时时间（避免空闲连接占用资源）
          idle-timeout: 300s # 空闲超时时间（5分钟无数据则断开）
        client:
          response-timeout: 60s # 客户端响应超时（可选）

# 日志配置（避免日志打印占用资源）
logging:
  level:
    root: INFO
    org.springframework.web: WARN
    reactor.netty: WARN # Netty 日志级别调高，减少输出

（2）JVM 参数优化（启动脚本）

# 示例：Linux 启动脚本（调整堆内存、线程栈、GC 算法）
java -jar sse-demo.jar 
  -Xms4G -Xmx4G  # 堆内存（根据服务器内存调整，避免过大导致 GC 卡顿）
  -Xss256K  # 线程栈大小（响应式模型线程少，可减小栈内存）
  -XX:+UseG1GC  # G1 GC 适合高并发场景，减少 STW 时间
  -XX:MaxGCPauseMillis=200  # 最大 GC 停顿时间（200ms）
  -XX:+ParallelRefProcEnabled  # 并行处理引用对象
  -XX:+UnlockExperimentalVMOptions 
  -XX:+UseCGroupMemoryLimitForHeap # 适配容器内存限制（Docker/K8s 环境）

（3）服务器内核参数优化（Linux）

调整 Linux 内核参数，支持更多文件描述符（每个连接对应一个文件描述符）：

# 临时生效（重启服务器失效）
ulimit -n 655350 # 单个进程最大文件描述符（设置为 65 万+）
sysctl -w net.core.somaxconn=65535 # 最大半连接队列大小
sysctl -w net.ipv4.tcp_max_syn_backlog=65535 # TCP 连接队列大小
sysctl -w net.ipv4.tcp_tw_reuse=1 # 复用 TIME_WAIT 状态的连接
sysctl -w net.ipv4.tcp_fin_timeout=30 # TIME_WAIT 超时时间（30秒）

# 永久生效（修改 /etc/sysctl.conf）
cat >> /etc/sysctl.conf << EOF
net.core.somaxconn=65535
net.ipv4.tcp_max_syn_backlog=65535
net.ipv4.tcp_tw_reuse=1
net.ipv4.tcp_fin_timeout=30
EOF
sysctl -p # 生效配置

# 修改用户文件描述符限制（/etc/security/limits.conf）
cat >> /etc/security/limits.conf << EOF
* soft nofile 655350
* hard nofile 655350
EOF

三、性能压测：验证 10 万 + 并发

1. 压测工具选择

用 JMeter + 自定义 SSE 压测脚本（或用 wrk 结合 Lua 脚本），模拟 10 万 + 并发连接。

（1）JMeter 压测脚本配置

添加「线程组」：设置线程数 100000， ramp-up 时间 60 秒（逐步加压，避免瞬间雪崩）；
添加「HTTP 请求」：请求方法 GET，URL http://localhost:8080/sse/connect/${userId}（userId 用随机函数生成）；
添加「响应断言」：验证响应头 Content-Type 为 text/event-stream；
添加「监听器」：聚合报告、查看结果树、服务器监控（CPU / 内存 / 连接数）。

2. 压测结果预期（4C8G 服务器）

指标	结果
并发连接数	100000+
平均响应时间	< 50ms
服务器 CPU 使用率	< 70%
内存占用	< 3GB（堆内存）
消息推送延迟	< 100ms（单条消息）
连接稳定性	持续 1 小时无断开

3. 压测注意事项

逐步加压：ramp-up 时间设置为 60-120 秒，避免瞬间创建 10 万连接导致服务器端口耗尽；
关闭日志：压测时将日志级别调至 ERROR，避免日志 IO 成为瓶颈；
监控内核参数：用 netstat -an | grep 8080 | wc -l 查看实际连接数，用 top 监控 CPU / 内存。

四、进阶优化：支撑更高并发

1. 连接分片（集群部署）

单机并发上限受服务器资源限制（文件描述符、内存），集群部署时需解决「用户连接路由」问题：

方案 1：会话粘滞（Sticky Session）：通过 Nginx 配置 ip_hash，将同一用户的连接路由到同一台服务器；
方案 2：分布式连接管理：用 Redis Pub/Sub 或 RocketMQ 实现跨节点消息广播（用户连接仍在本地，跨节点消息通过中间件转发）。

Nginx 会话粘滞配置（方案 1）

http {
    upstream sse_servers {
        server 192.168.1.100:8080;
        server 192.168.1.101:8080;
        ip_hash; # 基于客户端 IP 路由到固定服务器
    }

    server {
        listen 80;
        server_name sse-demo.com;

        location /sse/ {
            proxy_pass http://sse_servers;
            proxy_set_header Host $host;
            proxy_set_header X-Real-IP $remote_addr;
            # 关键：启用 HTTP 长连接支持
            proxy_http_version 1.1;
            proxy_set_header Connection "";
            proxy_read_timeout 300s; # 与服务端 idle-timeout 一致
        }
    }
}

2. 消息批量推送（减少 IO 开销）

如果需要向大量用户推送一样消息，可批量打包消息（如每 100ms 批量发送一次），减少 FluxSink 的 next() 调用次数：

// 批量广播优化（定时批量发送）
@Scheduled(fixedRate = 100) // 每 100ms 批量发送一次
public void batchBroadcast() {
    if (messageQueue.isEmpty() || userSinks.isEmpty()) return;
    // 批量获撤销息（假设 messageQueue 是线程安全的队列）
    List<String> messages = new ArrayList<>(messageQueue);
    messageQueue.clear();
    if (messages.isEmpty()) return;

    // 合并消息（按 SSE 格式拼接）
    String batchMessage = String.join("
", messages);
    // 广播批量消息
    broadcast(batchMessage);
}

3. 连接认证与权限控制

实际场景中需验证用户合法性，避免恶意连接占用资源：

@GetMapping(value = "/connect/{userId}", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<ServerSentEvent<String>> connect(
        @PathVariable String userId,
        @RequestHeader("Authorization") String token) {
    // 1. 验证 token 合法性（如解析 JWT）
    boolean isValid = authService.validateToken(token, userId);
    if (!isValid) {
        // 返回 401 未授权（Flux.error 触发异常响应）
        return Flux.error(new ResponseStatusException(HttpStatus.UNAUTHORIZED, "token 无效"));
    }
    // 2. 验证用户权限（如是否有权限接收该类型推送）
    boolean hasPermission = permissionService.checkSsePermission(userId);
    if (!hasPermission) {
        return Flux.error(new ResponseStatusException(HttpStatus.FORBIDDEN, "无推送权限"));
    }
    // 3. 正常建立连接
    return Flux.create(sink -> {
        connectionManager.register(userId, sink);
        // ... 发送欢迎消息
    });
}