Spring SseEmitter 系统及作用详细讲解_Java

springsseemitter系统详细讲解

一、sse 基本概念（server-sent events）

sse 是 html5 提出的标准。
客户端通过 eventsource 向服务端发起 一个长连接，服务器通过该连接持续向客户端发送事件。
相比 websocket（双向通信），sse 是单向的：服务端 → 客户端。
本质上是基于 http 协议的长连接。

示例（前端代码）

const eventsource = new eventsource("/sse/subscribe");
eventsource.onmessage = function(event) {
    console.log("收到消息:", event.data);
};
eventsource.onerror = function(e) {
    console.error("连接错误", e);
};

二、spring 的sseemitter作用

spring 提供了 sseemitter 类来简化 sse 的开发，它本质上是一个 controller 的返回对象，用于不断地往前端推送数据。

三、基本用法

1. 添加依赖（spring boot web 项目一般已包含）

<!-- maven 中无需额外添加，spring-boot-starter-web 已包含 -->

2. 控制器定义 sse 接口

@restcontroller
public class ssecontroller {
    @getmapping("/sse/subscribe")
    public sseemitter subscribe() {
        sseemitter emitter = new sseemitter(0l); // 不超时，或设置时间，如30_000l
        executors.newsinglethreadexecutor().submit(() -> {
            try {
                for (int i = 1; i <= 5; i++) {
                    emitter.send("第 " + i + " 条消息");
                    thread.sleep(1000);
                }
                emitter.complete(); // 关闭连接
            } catch (exception e) {
                emitter.completewitherror(e);
            }
        });
        return emitter;
    }
}

四、sseemitter 的重要方法

方法	说明
`send(object data)`	向客户端发送消息，支持字符串、json 等
`send(object data, mediatype mediatype)`	指定 mime 类型发送
`complete()`	正常关闭连接
`completewitherror(throwable ex)`	异常关闭连接
`oncompletion(runnable callback)`	设置连接关闭的回调
`ontimeout(runnable callback)`	设置超时回调
`onerror(consumer<throwable> callback)`	设置错误回调
`settimeout(long timeout)`	设置连接超时时间（默认 30 秒）

五、注意事项

1. content-type

content-type: text/event-stream

spring 会自动设置，只需：

@getmapping(value = "/sse/subscribe", produces = mediatype.text_event_stream_value)

2. 浏览器自动重连机制

如果服务端关闭连接（未调用 .complete()），浏览器会默认自动尝试重新连接。
禁止客户端重连可发送 retry: 0。

3. 多用户支持

private final map<string, sseemitter> useremitters = new concurrenthashmap<>();
@getmapping("/subscribe/{userid}")
public sseemitter subscribe(@pathvariable string userid) {
    sseemitter emitter = new sseemitter(60_000l);
    useremitters.put(userid, emitter);
    emitter.oncompletion(() -> useremitters.remove(userid));
    emitter.ontimeout(() -> useremitters.remove(userid));
    return emitter;
}
public void sendtouser(string userid, string message) throws ioexception {
    sseemitter emitter = useremitters.get(userid);
    if (emitter != null) {
        emitter.send(message);
    }
}

六、典型应用场景

实时进度条
实时通知系统
日志推送
后台任务结果反馈

七、sseemitter 与 websocket 的对比

比较项	sse（sseemitter）	websocket
通信方向	单向（服务端 → 客户端）	双向
协议	http	ws:// 或 wss://
浏览器支持	广泛支持	广泛支持
实现复杂度	简单	需要管理连接
使用场景	实时推送	聊天、游戏等

八、sseemitter 高级技巧

✅ 自定义事件名称

emitter.send(sseemitter.event()
    .name("customevent")
    .data("这是自定义事件")
);

前端监听：

eventsource.addeventlistener("customevent", function(event) {
    console.log("收到自定义事件:", event.data);
});

✅ 定时心跳保持连接

scheduledexecutorservice heartbeatscheduler = executors.newscheduledthreadpool(1);
heartbeatscheduler.scheduleatfixedrate(() -> {
    try {
        emitter.send("heartbeat");
    } catch (ioexception e) {
        emitter.completewitherror(e);
    }
}, 0, 15, timeunit.seconds);