SSE示例

SSE 是一种通过 HTTP 协议实现的服务器向客户端单向推送数据的标准技术。允许服务器主动向浏览器推送文本数据流，而不需要客户端频繁发起请求（即无需轮询）。可应用于消息推送、实时通知、系统告警等场景。

# webflux 中使用

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

@Slf4j
@RestController
@RequestMapping(value = "/test")
@CrossOrigin("*")
public class TestController {

    @GetMapping(path = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<String> stream() {
        return Flux.interval(Duration.ofSeconds(1))
                .map(l -> "Message " + count.getAndIncrement())
                .doOnCancel(() -> {
                    // 当连接关闭时，可以执行一些操作，比如重置计数
                    count.set(0);
                    System.out.println("链接关闭");
                });
    }

}

# wevmvc中使用

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


@Slf4j
@RestController
@RequestMapping(value = "/test")
@CrossOrigin("*")
public class SseController {

    @GetMapping("/stream")
    public SseEmitter stream() {
        SseEmitter emitter = new SseEmitter();

        // 使用 ScheduledExecutorService 每秒发送一条消息
        ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
        executor.scheduleAtFixedRate(() -> {
            try {
                // 发送消息
                emitter.send("Message at " + System.currentTimeMillis());
            } catch (IOException e) {
                emitter.completeWithError(e); // 出错时完成 emitter
                executor.shutdown(); // 停止执行器
            }
        }, 0, 1, TimeUnit.SECONDS);

        // 设置超时（可选）
        emitter.onTimeout(() -> {
            emitter.complete(); // 超时完成 emitter
            executor.shutdown(); // 停止执行器
        });

        // 处理连接关闭
        emitter.onCompletion(() -> {
            executor.shutdown(); // 停止执行器
        });

        return emitter;
    }
}

# 网页示例

点击查看

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>SSE Example</title>
    <style>
        body {
            font-family: Arial, sans-serif;
        }
        #messages {
            margin-top: 20px;
            padding: 10px;
            border: 1px solid #ccc;
            max-height: 300px;
            overflow-y: auto;
        }
        #closeButton {
            margin-top: 20px;
        }
    </style>
</head>
<body>
<h1>Server-Sent Events Example</h1>
<div id="messages"></div>

<button id="closeButton">关闭连接</button> <!-- 添加关闭按钮 -->
<button id="reconnect">重新连接</button> <!-- 添加关闭按钮 -->

<script>
    let eventSource;
    const messagesDiv = document.getElementById('messages');

    function connect() {
        eventSource = new EventSource('http://127.0.0.1:8081/test/stream');

        // 处理接收到的消息
        eventSource.onmessage = function(event) {
            messagesDiv.innerHTML += `<div>${event.data}</div>`;
            messagesDiv.scrollTop = messagesDiv.scrollHeight; // 自动滚动到最新消息
        };

        // 处理错误
        eventSource.onerror = function(event) {
            console.error('EventSource failed:', event);
            eventSource.close(); // 关闭连接
        };
    }

    // 关闭连接的功能
    document.getElementById('reconnect').addEventListener('click', function() {
        messagesDiv.innerHTML += `<div>开始重连。</div>`;
        connect(); // 尝试重新连接
    });

    // 关闭连接的功能
    document.getElementById('closeButton').addEventListener('click', function() {
        eventSource.close();
        messagesDiv.innerHTML += `<div>连接已关闭。</div>`;
    });

    // 初始连接
    connect();
</script>
</body>
</html>

上次更新: 2025/04/25, 10:46:05

← 初探webflux Shell基础→