SSE（Server Sent Event）实战（2）- Spring MVC 实现

news2025/2/22 5:55:55

一、服务端实现

使用 @RestController 注解创建一个控制器类（Controller）
创建一个方法来创建一个客户端连接，它返回一个 SseEmitter，处理 GET 请求并产生（produces）文本/事件流 (text/event-stream)
创建一个新的 SseEmitter, 保存它并从方法中返回
在另一个线程中异步发送事件, 先拿到保存的 SseEmitter 并根据需要多次调用调用SseEmitter.send()方法
完成事件发送, 调用 SseEmitter.complete() 方法
要异常完成发送事件，请调用 SseEmitter.completeWithError() 方法

/*
 * xxx.com
 * Copyright (C) 2021-2024 All Rights Reserved.
 */
package com.sse.demo.controller;

import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

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

/**
 * @author xxx
 * @version SseController.java, v 0.1 2024-07-11 10:11
 */
@Slf4j
@RestController
@RequestMapping("/sse")
public class SseController {

    private static final Map<String, SseEmitter> SSE_EMITTER_MAP = new ConcurrentHashMap<>();

    /**
     * 创建连接
     */
    @GetMapping("/create-connect")
    public SseEmitter createConnect(@RequestParam("userId") String userId) {

        try {
            // 设置超时时间，0表示不过期。默认30秒
            SseEmitter sseEmitter = new SseEmitter(0L);

            // 注册回调
            sseEmitter.onCompletion(() -> removeSseConnection(userId, "SSE连接已关闭"));
            sseEmitter.onError(throwable -> removeSseConnection(userId, "SSE连接出现错误"));
            sseEmitter.onTimeout(() -> removeSseConnection(userId, "SSE连接超时"));

            SSE_EMITTER_MAP.put(userId, sseEmitter);

            log.info("创建了用户[{}]的SSE连接", userId);
            return sseEmitter;
        } catch (Exception e) {
            log.error("创建新的SSE连接异常，当前用户：" + userId, e);
            return null;
        }
    }

    /**
     * 发送消息
     */
    @GetMapping("/send-message")
    public void sendMessage(@RequestParam("userId") String userId, @RequestParam("message") String message) {

        SseEmitter sseEmitter = SSE_EMITTER_MAP.get(userId);
        if (sseEmitter != null) {
            try {
                sseEmitter.send(SseEmitter.event()
                        .name("message")
                        .data(message)
                        .reconnectTime(5000));
                log.info("给用户[{}]发送消息成功: {}", userId, message);
            } catch (Exception e) {
                log.error("给用户[{}]发送消息失败: {}", userId, e.getMessage(), e);
                // 如果发送失败，尝试从map中移除失效的SseEmitter
                removeSseConnection(userId, "发送消息失败");
            }
        } else {
            log.info("用户[{}]的SSE连接不存在或已关闭，无法发送消息", userId);
        }
    }

    private void removeSseConnection(String userId, String reason) {
        SSE_EMITTER_MAP.computeIfPresent(userId, (key, sseEmitter) -> {
            sseEmitter.complete();
            log.info("用户[{}]的SSE连接已移除，原因：{}", userId, reason);
            return null;
        });
    }
}

二、客户端实现

创建多个 index.html文件，放在 static 目录下，用不同的浏览器打开，实现向多个用户推送的场景。
在这里插入图片描述


<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>SSE Demo</title>
    <script>        document.addEventListener('DOMContentLoaded', function () {
        var userId = "1";

        // 创建一个新的EventSource对象
        var source = new EventSource('http://localhost:8080/sse/create-connect?userId=' + userId);

        // 当连接打开时触发
        source.onopen = function (event) {
            console.log('SSE连接已打开');
        };

        // 当从服务器接收到消息时触发
        source.onmessage = function (event) {
            // event.data 包含服务器发送的文本数据
            console.log('接收到消息:', event.data);
            // 在页面上显示消息
            var messagesDiv = document.getElementById('messages');
            if (messagesDiv) {
                messagesDiv.innerHTML += '<p>' + event.data + '</p>'; // 直接使用event.data
            } else {
                console.error('未找到消息容器元素');
            }
        };

        // 当发生错误时触发
        source.onerror = function (event) {
            console.error('SSE连接错误:', event);
        };
    });
    </script>
</head>
<body>
<div id="messages">
    <!-- 这里将显示接收到的消息 -->
</div>
</body>
</html>