Web服务端通过SSE推送消息给浏览器客户端的实现方案(附详细代码和仓库地址)

- 1、SSE（Server-Sent Events）简介
- 2、SSE 的工作原理
- 3、SSE 与客户端轮询的区别和优势比较
- - 区别
  - 优势
- 4、SSE简单实现(单机应用Demo)
- - 演示效果
  - SSE-Demo仓库地址
  - 下面直接贴代码：
  - 前端实现：
  - 后端实现：
- 5、SSE简单实现(分布式应用Demo)
- - SSE-Demo仓库地址
  - 关键代码
  - 方案说明

1、SSE（Server-Sent Events）简介

Server-Sent Events (SSE) 是一种基于 HTTP 协议的服务器推送技术，允许服务器通过单个持久连接向客户端发送实时更新。客户端使用标准的 EventSource API 来接收服务器推送的事件，这种通信方式非常适合实时应用，如消息通知、股票行情更新、社交媒体更新等。

2、SSE 的工作原理

单向连接：SSE 建立的是单向通道，即服务器向客户端推送数据，客户端只能接收，不能发送。
持久连接：SSE 使用的是长连接（Long Polling），即连接一旦建立，将会持续存在，直到客户端或服务器关闭连接。
文本数据：SSE 通过 text/event-stream MIME 类型传输数据，数据是纯文本格式。
自动重连：如果连接中断，EventSource 会自动尝试重新连接，确保客户端能够接收后续的推送。

3、SSE 与客户端轮询的区别和优势比较

客户端轮询（Client Polling）是一种传统的客户端从服务器请求数据的方式。客户端会定期向服务器发送请求，检查是否有新数据可用。

区别

连接方式：
SSE：建立后服务器主动推送数据，连接是持久的，数据在有更新时实时传递。
轮询：客户端定期发送请求获取数据，连接是间歇性的。

实时性：
SSE：数据几乎是实时推送的，延迟极低。
轮询：数据获取延迟取决于轮询的频率，频率高则延迟低，但频率低可能导致数据延迟。

网络和服务器负载：
SSE：由于是单个持久连接，减少了频繁的请求与响应开销，降低了服务器负载。
轮询：频繁的请求会增加服务器和网络的负担，尤其是在轮询频率较高时。

连接控制：
SSE：自动处理连接中断和重连，客户端实现简单。
轮询：需要客户端定期发起请求，且如果请求频率不当，可能导致资源浪费。

数据传输效率：
SSE：只在有数据更新时推送，传输效率高。
轮询：即使没有数据更新，客户端也会定期请求，效率低下。

优势

SSE 的优势：
更高效的网络和服务器资源利用率。
实时性更高，延迟更低。
实现简单，特别是在浏览器环境中，支持自动重连和事件处理。
适合需要频繁更新但客户端无需响应的场景。

客户端轮询的优势：
在不支持 SSE 的环境下仍然可以使用。
实现和理解相对简单，兼容性更好。

4、SSE简单实现(单机应用Demo)

演示效果

在这里插入图片描述

SSE-Demo仓库地址

https://github.com/deepjava-gm/SSE-Demo.git

下面直接贴代码：

前端实现：

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>SSE 用户消息推送 Demo</title>
    <style>
        body {
            font-family: Arial, sans-serif;
            margin: 20px;
        }
        h1 {
            text-align: center;
        }
        .container {
            max-width: 600px;
            margin: 0 auto;
            padding: 20px;
            border: 1px solid #ccc;
            border-radius: 8px;
        }
        .form-group {
            margin-bottom: 15px;
        }
        label {
            display: block;
            margin-bottom: 5px;
            font-weight: bold;
        }
        input[type="text"] {
            width: 100%;
            padding: 8px;
            box-sizing: border-box;
            border: 1px solid #ccc;
            border-radius: 4px;
        }
        button {
            padding: 10px 20px;
            color: white;
            background-color: #007bff;
            border: none;
            border-radius: 4px;
            cursor: pointer;
        }
        button:hover {
            background-color: #0056b3;
        }
        #messages {
            margin-top: 20px;
        }
        .message {
            background-color: #f1f1f1;
            padding: 10px;
            margin-bottom: 10px;
            border-radius: 4px;
        }
        .status {
            margin-top: 10px;
            font-weight: bold;
        }
        .success {
            color: green;
        }
        .error {
            color: red;
        }
    </style>
</head>
<body>
    <h1>SSE 用户消息推送 Demo</h1>
    <div class="container">
        <!-- 连接部分 -->
        <div class="form-group">
            <label for="userId">用户 ID:</label>
            <input type="text" id="userId" placeholder="请输入您的用户 ID">
            <br><br>
            <button onclick="connect()">连接</button>
        </div>
        <div id="connectionStatus" class="status"></div>

        <!-- 消息发送部分 -->
        <div class="form-group">
            <label for="targetUserId">目标用户 ID:</label>
            <input type="text" id="targetUserId" placeholder="请输入目标用户 ID">
        </div>
        <div class="form-group">
            <label for="message">消息内容:</label>
            <input type="text" id="message" placeholder="请输入要发送的消息">
        </div>
        <button onclick="sendMessage()">推送消息</button>

        <!-- 消息显示部分 -->
        <div id="messages"></div>
    </div>

    <script>
        let eventSource;
        let currentUserId = '';

        function connect() {
            const userId = document.getElementById('userId').value;
            const connectionStatus = document.getElementById('connectionStatus');

            if (!userId) {
                alert('请输入用户 ID');
                return;
            }

            // 显示连接状态为“连接中”
            connectionStatus.textContent = '已连接...';
            connectionStatus.className = 'status';

            eventSource = new EventSource(`http://localhost:9999/sse/connect/${userId}`);
            currentUserId = userId; // 保存当前用户 ID

            eventSource.onopen = function() {
                connectionStatus.textContent = '接收成功';
                connectionStatus.className = 'status success';
            };

            eventSource.onmessage = function(event) {
                try {
                    // 解析 JSON 消息
                    const data = JSON.parse(event.data);
                    const newElement = document.createElement('div');
                    newElement.className = 'message';
                    newElement.innerText = `用户 ${data.senderId} 接收的消息: ${data.message}`;
                    document.getElementById('messages').appendChild(newElement);
                } catch (e) {
                    console.error('消息解析错误:', e);
                }
            };

            eventSource.onerror = function(event) {
                connectionStatus.textContent = '连接失败，请检查网络或服务器';
                connectionStatus.className = 'status error';
                console.error("连接错误: ", event);
                eventSource.close();
            };
        }

        function sendMessage() {
            const targetUserId = document.getElementById('targetUserId').value;
            const message = document.getElementById('message').value;
            
            if (!targetUserId || !message) {
                alert('请填写目标用户 ID 和消息内容');
                return;
            }

            // 发送 GET 请求推送消息
            fetch(`http://localhost:9999/sse/push/${targetUserId}?message=${encodeURIComponent(message)}`, {
                method: 'GET'
            }).then(response => {
                if (response.ok) {
                    console.log('消息发送成功');
                } else {
                    console.log('消息发送失败');
                }
            }).catch(error => {
                console.error('发送错误:', error);
                alert('消息发送失败');
            });
        }
    </script>
</body>
</html>

后端实现：

启动类：

package io.github.deepjava;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.util.concurrent.ConcurrentHashMap;

@SpringBootApplication
public class SseDemoApplication {
    public static void main(String[] args) {
        SpringApplication.run(SseDemoApplication.class);
    }


    // 注入一个全局缓存 用来保存不同用户的SSE连接信息
    @Bean("userSSEMap")
    public ConcurrentHashMap<String, SseEmitter> getUserSSEMap(){
        return new ConcurrentHashMap<>();
    }

}

Controller：

package io.github.deepjava.controller;

import io.github.deepjava.service.SseService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import javax.annotation.Resource;
import java.util.concurrent.ConcurrentHashMap;

@RestController
@RequestMapping("/sse")
@CrossOrigin(origins = "*")
@Slf4j
public class SseController {

    @Resource(name = "userSSEMap")
    private ConcurrentHashMap<String, SseEmitter> userSSEMap;

    @Resource
    private SseService sseService;

    // 连接方法：为用户 ID 注册 SSE 链接
    @GetMapping("/connect/{userId}")
    public SseEmitter connect(@PathVariable String userId) {
        SseEmitter emitter = new SseEmitter(0L); // 设置超时时间为无限大
        userSSEMap.put(userId, emitter);
        // 连接正常关闭回调 移除连接
        emitter.onCompletion(() -> {
            userSSEMap.remove(userId);
            log.info("连接正常关闭回调 移除连接");
        });
        // 连接超时回调 移除连接
        emitter.onTimeout(() -> {
            userSSEMap.remove(userId);
            log.info("连接超时回调 移除连接");

        });
        // 连接出错回调 移除连接
        emitter.onError((e) -> {
            userSSEMap.remove(userId);
            log.info("连接出错回调 移除连接");
        });
        log.info("连接成功！");
        return emitter;
    }

    // 推送方法：根据用户 ID 发送消息
    @GetMapping("/push/{userId}")
    public void push(@PathVariable String userId, @RequestParam String message) {
        sseService.extracted(userId, message);
    }

}

Service：

package io.github.deepjava.service;


import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.concurrent.ConcurrentHashMap;

@Service
@Slf4j
public class SseService {

    @Resource(name = "userSSEMap")
    private ConcurrentHashMap<String, SseEmitter> clients;

    public void extracted(String userId, String message) {
        SseEmitter emitter = clients.get(userId);
        if (emitter != null) {
            try {
                // 创建包含用户 ID 和消息内容的 JSON 对象
                String jsonMessage = String.format("{\"senderId\":\"%s\", \"message\":\"%s\"}", userId, message);
                emitter.send(jsonMessage);
                log.info("消息推送成功！");
            } catch (IOException e) {
                clients.remove(userId);
                log.info("消息推送失败！");
            }
        }
    }

}

配置文件：application.properties

spring.application.name=sse-demo
server.port=9999

Maven的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>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.7.1</version>
    </parent>

    <groupId>org.example</groupId>
    <artifactId>SSE-Demo</artifactId>
    <version>1.0-SNAPSHOT</version>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <!--lombok  -->
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
    </dependencies>


    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>

</project>

5、SSE简单实现(分布式应用Demo)

注意：
SSE 连接（如 SseEmitter）是持久化的、与具体服务器实例相关联的动态对象，无法直接存储在 Redis 等外部存储中。Redis 主要用于消息传递和共享数据，但无法直接管理活跃的连接。所以下面方案仅使用 Redis 进行消息广播。

解决方案概述
为了在分布式环境中实现 SSE，通常采用以下架构：

每个服务器实例维护本地的 SSE 连接：每个实例只管理与自身连接的客户端。
使用 Redis 进行消息广播：当需要向特定用户推送消息时，将消息发布到 Redis 频道。所有实例订阅该频道，并检查自己是否有需要向某个用户推送的连接。
用户与实例的映射：使用 Redis 存储用户与服务器实例的映射信息，确保消息能够被正确路由到处理该用户连接的实例。
虽然无法完全将连接信息存储在 Redis 中，但通过这种方式，可以有效地在分布式环境中管理 SSE 连接和消息推送。

这里只贴主要的后端代码：完整代码去下载仓库代码看。

SSE-Demo仓库地址

https://github.com/deepjava-gm/SSE-Demo.git

关键代码

redis配置：

 // 配置redis的序列化
    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {
        RedisTemplate<String, Object> template = new RedisTemplate<>();
        template.setConnectionFactory(factory);
        template.setKeySerializer(new StringRedisSerializer());
        template.setValueSerializer(new StringRedisSerializer());
        return template;
    }


    @Bean
    public RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory connectionFactory) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        return container;
    }

redis监听主题：

package io.github.deepjava.controller;

import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.ConcurrentHashMap;

@RestController
@RequestMapping("/dis/sse")
@CrossOrigin(origins = "*")
@Slf4j
public class DistributedSseController {

    @Resource(name = "userSSEMap")
    private ConcurrentHashMap<String, SseEmitter> userSSEMap;

    @Resource
    private RedisTemplate<String, String> redisTemplate;

    @Resource
    private RedisMessageListenerContainer redisMessageListenerContainer;

    private final ChannelTopic topic = new ChannelTopic("sse-messages");

    @PostConstruct
    public void init() {
        // 订阅 Redis 频道
        redisMessageListenerContainer.addMessageListener(
                new MessageListenerAdapter((MessageListener) (message, pattern) -> {
                    String payload = new String(message.getBody(), StandardCharsets.UTF_8);

                    // 假设消息的格式为 "userId:message"
                    String[] parts = payload.split(":", 2);
                    if (parts.length == 2) {
                        String userId = parts[0];
                        String userMessage = parts[1];
                        // 发送消息给本地的 SSE 连接
                        SseEmitter emitter = userSSEMap.get(userId);
                        if (emitter != null) {
                            try {
                                String jsonMessage = String.format("{\"senderId\":\"%s\", \"message\":\"%s\"}", userId, userMessage);
                                emitter.send(jsonMessage);
                            } catch (IOException e) {
                                emitter.completeWithError(e);
                                userSSEMap.remove(userId);
                            }
                        }
                    }
                }), topic);
    }

    // 连接方法：为用户 ID 注册 SSE 链接
    @GetMapping("/connect/{userId}")
    public SseEmitter connect(@PathVariable String userId) {
        SseEmitter emitter = new SseEmitter(0L); // 设置超时时间为无限大
        userSSEMap.put(userId, emitter);
        // 连接正常关闭回调 移除连接
        emitter.onCompletion(() -> {
            userSSEMap.remove(userId);
            log.info("连接正常关闭回调 移除连接");
        });
        // 连接超时回调 移除连接
        emitter.onTimeout(() -> {
            userSSEMap.remove(userId);
            log.info("连接超时回调 移除连接");

        });
        // 连接出错回调 移除连接
        emitter.onError((e) -> {
            userSSEMap.remove(userId);
            log.info("连接出错回调 移除连接");
        });
        log.info("连接成功！");
        return emitter;
    }


    @GetMapping("/push/{userId}")
    public void push(@PathVariable String userId, String message) {
        // 将消息发布到 Redis 频道
        redisTemplate.convertAndSend(topic.getTopic(), userId + ":" + message);
    }
}