SpringBoot3.4.3基于Spring WebFlux实现SSE功能

itbeien2025年3月3日大约 9 分钟

SpringBoot3.4.3基于Spring WebFlux实现SSE功能

本专栏前一篇文章SpringBoot3.4.3基于Caffeine实现本地缓存。今天分享SpringBoot3.4.3基于Spring WebFlux实现SSE功能，开发者可以基于SSE技术开发有事件驱动场景的业务需求。

完整代码在文章最后，如果觉得本篇文章对你有用，记得点赞、关注、收藏哦。你的支持是我持续更新的动力!

文章最后可以加入免费的Java&AI技术和支付系统沟通社群，一起探讨Java/你的产品如何与AI结合，请按照要求加入。在群中可以聊开发、系统设计、架构、行业趋势、AI等等话题

SpringBoot3专栏软件环境

JDK17.0.12
SpringBoot3.4.3
IDEA2024.3.3
spring-boot-starter-webflux

我们先看本篇文章对应的项目结构，请看下图

Server-Sent Events（SSE）是一种允许服务器向客户端发送实时更新的 Web 技术。下面为你详细介绍 SSE 的定义、工作原理、优点及使用场景

1 什么是SSE及原理

SSE 基于 HTTP 协议，允许服务器在客户端建立连接后，持续不断地向客户端发送事件流。客户端使用 EventSource 对象（在浏览器环境中）来建立与服务器的连接，并监听服务器发送的事件。一旦连接建立，服务器可以随时向客户端发送新的数据，而无需客户端主动发起请求。

1.1 优点

1.1.1 简单易用

技术层面：SSE 基于 HTTP 协议，其实现相对简单，不需要像 WebSocket 那样复杂的握手和协议处理。对于开发者来说，只需要在服务器端设置好响应头，以 text/event-stream 格式发送数据，客户端使用 EventSource API 就可以轻松接收数据。
代码层面：在服务器端，使用 Spring Boot 等框架可以很方便地实现 SSE 功能，几行代码就可以实现一个简单的 SSE 服务。在客户端，使用 JavaScript 的 EventSource 对象也非常直观，只需要处理 onmessage 和 onerror 等事件即可。

1.1.2 自动重连

机制：当客户端与服务器之间的连接意外中断时，EventSource 对象会自动尝试重新建立连接。而且，它会在重连时发送一个 Last - Event - ID 头，告诉服务器从哪个事件开始继续发送数据，确保数据的连续性。
应用场景：在网络不稳定的环境中，如移动网络场景下，自动重连机制可以保证客户端能够尽可能及时地恢复数据接收，减少数据丢失。

1.1.3 基于 HTTP 协议

兼容性：由于 SSE 基于 HTTP 协议，它可以直接通过现有的 HTTP 基础设施进行传输，因此具有很好的兼容性。几乎所有现代浏览器都支持 EventSource API，无需额外的插件或配置。
安全性：可以利用现有的 HTTP 安全机制，如 HTTPS 进行加密传输，保障数据的安全性。

1.1.4 事件驱动

数据推送方式：SSE 是一种事件驱动的通信方式，服务器可以根据业务逻辑在合适的时机向客户端推送数据。这种方式非常适合实时数据更新的场景，如推送新消息、实时推送系统监控等。
解耦性：服务器和客户端之间的耦合度较低，服务器只需要关注何时发送事件，而客户端只需要关注如何处理接收到的事件，双方可以独立开发和维护。

1.1.5 服务器资源占用相对较低

连接方式：与一些长轮询或其他实时通信方式相比，SSE 只需要建立一个持久的 HTTP 连接，服务器不需要为每个请求创建新的连接，从而减少了服务器的资源消耗。
带宽利用：在数据传输方面，SSE 采用文本格式，并且可以根据实际情况控制数据发送的频率，有效利用带宽资源。

2 使用场景

实时通知和消息提醒：实时推送新消息、通知或警报给用户。例如：业务系统中的的新邮件提醒、社交平台的消息提醒。
监控与日志流：实时推送系统监控指标、错误日志等信息到前端页面，便于运维人员及时发现并处理问题。
物联网（IoT）设备状态更新：实时推送连接到网络的物理设备的状态变化，如传感器读数、设备健康状况等。
其它企业级应用系统

3 项目实战

3.1 pom依赖

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>cn.itbeien</groupId>
        <artifactId>springboot3-labs-master</artifactId>
        <version>1.0-SNAPSHOT</version>
    </parent>

    <artifactId>springboot-sse</artifactId>

    <properties>
        <maven.compiler.source>17</maven.compiler.source>
        <maven.compiler.target>17</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    </properties>

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

</project>

3.2 代码实现

package cn.itbeien.sse.controller;

import org.springframework.http.MediaType;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;

import java.time.Duration;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author itbeien
 * 项目网站：https://www.itbeien.cn
 * 公众号：贝恩聊架构
 * 全网同名，欢迎小伙伴们关注
 * Java/AI学习社群
 * Copyright© 2025 itbeien
 */
@RestController
public class SSEController {
    // 定义一个AtomicInteger类型的计数器
    private final AtomicInteger counter = new AtomicInteger(0);

    // 定义一个GET请求，返回ServerSentEvent类型的Flux
    @GetMapping(path = "/sse", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<Integer>> streamEvents() {
        // 每隔1秒返回一个ServerSentEvent，数据为计数器的值
        return Flux.interval(Duration.ofSeconds(1))
                .map(seq -> ServerSentEvent.<Integer>builder()
                        .data(counter.incrementAndGet())
                        .build());
    }
}