Spring Boot实战：5分钟搞定SSE消息推送（含完整代码示例）

Spring Boot实战：5分钟构建股票行情推送系统（SSE全流程指南）

1. 为什么选择SSE技术？

在实时数据推送领域，开发者常面临技术选型的困惑。当我们需要实现股票行情更新这类高频单向数据推送场景时，Server-Sent Events（SSE）往往是最优雅的解决方案。与传统的轮询和复杂的WebSocket相比，SSE具备三大核心优势：

1. 极简实现：基于标准HTTP协议，无需额外协议栈
2. 自动恢复：内置连接中断重试机制
3. 资源高效：单个连接支持持续数据流

// 传统轮询 vs SSE 请求对比 @GetMapping("/polling") // 传统轮询 public String getStockPrice() { return stockService.getLatestPrice(); // 每次请求都需建立新连接 } @GetMapping("/sse") // SSE public SseEmitter streamStockPrice() { return sseService.createEmitter(); // 单次连接持续推送 }

提示：SSE特别适合股票行情、新闻推送、实时监控等场景，这些场景中90%以上的数据流是服务器到客户端的单向传输

2. 快速搭建SSE服务端

2.1 基础环境配置

确保Spring Boot项目包含web依赖（Spring Boot 2.7+或3.0+）：

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

2.2 核心服务层实现

创建StockSSEService.java处理连接管理：

@Service @Slf4j public class StockSSEService { private final Map<String, SseEmitter> emitters = new ConcurrentHashMap<>(); public SseEmitter subscribe(String stockCode) { SseEmitter emitter = new SseEmitter(30 * 60 * 1000L); // 30分钟超时 emitter.onCompletion(() -> { log.info("{} SSE连接正常关闭", stockCode); emitters.remove(stockCode); }); emitter.onTimeout(() -> { log.warn("{} SSE连接超时", stockCode); emitters.remove(stockCode); }); emitters.put(stockCode, emitter); return emitter; } public void pushPriceUpdate(String stockCode, String price) { if (emitters.containsKey(stockCode)) { try { emitters.get(stockCode).send( SseEmitter.event() .id(UUID.randomUUID().toString()) .data(price) ); } catch (IOException e) { log.error("{} 推送失败: {}", stockCode, e.getMessage()); emitters.remove(stockCode); } } } }

2.3 控制器层设计

StockController.java暴露两个关键端点：

@RestController @RequestMapping("/api/stocks") public class StockController { @Autowired private StockSSEService sseService; @GetMapping("/subscribe/{code}") public SseEmitter subscribe(@PathVariable String code) { return sseService.subscribe(code); } @PostMapping("/mock-update/{code}") public ResponseEntity<?> mockUpdate( @PathVariable String code, @RequestParam String price) { sseService.pushPriceUpdate(code, price); return ResponseEntity.ok().build(); } }

3. 前端实时监听实现

3.1 基础事件监听

<div id="stock-ticker"> <h3>股票代码: <span id="stock-code"></span></h3> <div id="price-display">--</div> </div> <script> const stockCode = "AAPL"; // 示例股票代码 document.getElementById("stock-code").textContent = stockCode; const eventSource = new EventSource(`/api/stocks/subscribe/${stockCode}`); eventSource.onmessage = (event) => { document.getElementById("price-display").textContent = `$${parseFloat(event.data).toFixed(2)}`; flashPriceChange(); // 价格变化视觉反馈 }; eventSource.onerror = () => { console.error("SSE连接异常"); // 实现自动重连逻辑... }; </script>

3.2 增强型处理（带事件类型）

eventSource.addEventListener("priceUpdate", (e) => { updatePrice(e.data); }); eventSource.addEventListener("volumeAlert", (e) => { showAlert(`交易量异常: ${e.data}`); });

4. 生产环境进阶配置

4.1 性能优化参数

@Configuration public class SSEConfig implements WebMvcConfigurer { @Override public void configureAsyncSupport(AsyncSupportConfigurer configurer) { configurer.setDefaultTimeout(1800000); configurer.setTaskExecutor(taskExecutor()); } @Bean public ThreadPoolTaskExecutor taskExecutor() { ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); executor.setCorePoolSize(10); executor.setMaxPoolSize(50); executor.setQueueCapacity(100); return executor; } }

4.2 安全防护措施

1. 连接验证：

@GetMapping("/subscribe/{code}") public SseEmitter subscribe( @PathVariable String code, @RequestHeader("Authorization") String token) { if (!securityService.validateToken(token)) { throw new SecurityException("无效凭证"); } return sseService.subscribe(code); }

2. CORS配置：

@Bean public WebMvcConfigurer corsConfigurer() { return new WebMvcConfigurer() { @Override public void addCorsMappings(CorsRegistry registry) { registry.addMapping("/api/stocks/**") .allowedOrigins("https://yourdomain.com") .allowedMethods("GET"); } }; }

5. 典型问题排查指南

5.1 常见问题速查表

5.2 心跳机制实现

@Scheduled(fixedRate = 15000) public void sendHeartbeat() { emitters.forEach((code, emitter) -> { try { emitter.send(SseEmitter.event().comment("hb")); } catch (IOException e) { emitter.completeWithError(e); } }); }

6. 扩展应用场景

6.1 多股票订阅方案

public SseEmitter subscribeMultiple(@RequestParam List<String> codes) { SseEmitter emitter = new SseEmitter(); codes.forEach(code -> { StockWatcher watcher = new StockWatcher(code, price -> { try { emitter.send( SseEmitter.event() .name(code) .data(price) ); } catch (IOException ex) { // 错误处理 } }); stockService.registerWatcher(watcher); }); return emitter; }

6.2 与消息队列集成

@KafkaListener(topics = "stock-updates") public void handleStockUpdate(StockUpdate update) { if (emitters.containsKey(update.getCode())) { emitters.get(update.getCode()) .send(update.toSSEEvent()); } }

7. 性能压测数据参考

使用JMeter对SSE服务进行压力测试（单台4核8G服务器）：

关键优化建议：

• 使用HTTP/2减少连接开销
• 对非活跃连接实施清理策略
• 考虑分布式部署时使用Redis发布/订阅

8. 完整代码示例

8.1 服务端增强版

// 在StockSSEService中添加历史数据推送 public SseEmitter subscribeWithHistory(String code) { SseEmitter emitter = subscribe(code); executor.execute(() -> { try { // 推送最近10条历史数据 stockService.getHistoryPrices(code, 10) .forEach(price -> { emitter.send(price.toSSEEvent()); }); } catch (Exception e) { emitter.completeWithError(e); } }); return emitter; }

8.2 前端增强版

// 添加重连逻辑 function setupSSE() { const eventSource = new EventSource(url); eventSource.onerror = () => { eventSource.close(); setTimeout(setupSSE, 3000); // 3秒后重连 }; return eventSource; } // 添加数据缓存 const priceHistory = []; eventSource.onmessage = (e) => { priceHistory.push({ time: new Date(), price: e.data }); renderChart(priceHistory); };