JAVA通过SSE实现消息推送

2025-03-09 约 2490 字预计阅读 5 分钟

https://bing.ee123.net/img/rand?artid=146129940

JAVA通过SSE实现消息推送

1.什么是SSE？

SSE（Server-Sent Events）是一种用于实现服务器主动向客户端推送数据的技术，也被称为“事件流”（Event Stream）。它基于 HTTP 协议，利用了其长连接特性，在与服务器之间建立一条持久化连接，并通过这条连接实现服务器向客户端的实时数据推送。

2.SSE技术的基本原理

客户端向服务器发送一个GET请求，带有指定的header，表示可以接收事件流类型，并禁用任何的事件缓存。
服务器返回一个响应，带有指定的header，表示事件的媒体类型和编码，以及使用分块传输编码（chunked）来流式传输动态生成的内容。
服务器在有数据更新时，向客户端发送一个或多个名称：值字段组成的事件，由单个换行符分隔。事件之间由两个换行符分隔。服务器可以发送事件数据、事件类型、事件ID和重试时间等字段。
客户端使用EventSource接口来创建一个对象，打开连接，并订阅onopen、onmessage和onerror等事件处理程序来处理连接状态和接收消息。
客户端可以使用GET查询参数来传递数据给服务器，也可以使用close方法来关闭连接。

3.SSE和Socket的区别

SSE（Server-Sent Events）和 WebSocket 都是实现服务器向客户端实时推送数据的技术，但它们在某些方面还是有一定的区别。

技术实现

SSE 基于 HTTP 协议，利用了其长连接特性，通过浏览器向服务器发送一个 HTTP 请求，建立一条持久化的连接。而 WebSocket 则是通过特殊的升级协议（HTTP/1.1 Upgrade 或者 HTTP/2）建立新的 TCP 连接，与传统 HTTP 连接不同。

数据格式

SSE 可以传输文本和二进制格式的数据，但只支持单向数据流，即只能由服务器向客户端推送数据。WebSocket 支持双向数据流，客户端和服务器可以互相发送消息，并且没有消息大小限制。

连接状态

SSE 的连接状态仅有三种==：已连接、连接中、已断开==。连接状态是由浏览器自动维护的，客户端无法手动关闭或重新打开连接。而 WebSocket 连接的状态更灵活，可以手动打开、关闭、重连等。

兼容性

SSE 是标准的 Web API，可以在大部分现代浏览器和移动设备上使用。但如果需要兼容老版本的浏览器（如 IE6/7/8），则需要使用 polyfill 库进行兼容。而 WebSocket 在一些老版本 Android 手机上可能存在兼容性问题，需要使用一些特殊的 API 进行处理。

安全性

SSE 的实现比较简单，都是基于 HTTP 协议的，与普通的 Web 应用没有太大差异，因此风险相对较低。WebSocket 则需要通过额外的安全措施（如 SSL/TLS 加密）来确保数据传输的安全性，避免被窃听和篡改，否则可能会带来安全隐患。

总体来说，SSE 和 WebSocket 都有各自的优缺点，适用于不同的场景和需求。如果只需要服务器向客户端单向推送数据，并且应用在前端的浏览器环境中，则 SSE 是一个更加轻量级、易于实现和维护的选择。而如果需要双向传输数据、支持自定义协议、或者在更加复杂的网络环境中应用，则 WebSocket 可能更加适合。

SSE适用于场景

SSE适用场景是指服务器向客户端实时推送数据的场景，例如：

股票价格更新：服务器可以根据股市的变化，实时地将股票价格推送给客户端，让客户端能够及时了解股票的走势和行情。
新闻实时推送：服务器可以根据新闻的更新，实时地将新闻内容或标题推送给客户端，让客户端能够及时了解最新的新闻动态和信息。
在线聊天：服务器可以根据用户的发送，实时地将聊天消息推送给客户端，让客户端能够及时收到和回复消息。
实时监控：服务器可以根据设备的状态，实时地将监控数据或报警信息推送给客户端，让客户端能够及时了解设备的运行情况和异常情况。

SSE适用场景的特点是：

数据更新频繁：服务器需要不断地将最新的数据推送给客户端，保持数据的实时性和准确性。
低延迟：服务器需要尽快地将数据推送给客户端，避免数据的延迟和过期。
单向通信：服务器只需要向客户端推送数据，而不需要接收客户端的数据。

4.编写SSE服务，来进行创建链接和发送消息

Service:

package com.zillion.aggregate.app.controller;


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

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

@Slf4j
@Service
public class SSEService {

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

    public SseEmitter crateSse(String uid) {
        SseEmitter sseEmitter = new SseEmitter(0L);
        sseEmitter.onCompletion(() -> {
            log.info("[{}]结束链接" , uid);
            sseEmitterMap.remove(uid);
        });
        sseEmitter.onTimeout(() -> {
            log.info("[{}]链接超时",uid);
        });
        sseEmitter.onError(throwable -> {
            try{
                log.info("[{}]链接异常，{}",uid,throwable.toString());
                sseEmitter.send(SseEmitter.event()
                        .id(uid)
                        .name("发生异常")
                        .data("发生异常请重试")
                        .reconnectTime(3000));
                sseEmitterMap.put(uid,sseEmitter);
            }catch (IOException e){
                e.printStackTrace();
            }
        });
        try{
            sseEmitter.send(SseEmitter.event().reconnectTime(5000));
        }catch (IOException e){
            e.printStackTrace();
        }
        sseEmitterMap.put(uid,sseEmitter);
        log.info("[{}]创建sse连接成功!",uid);
        return sseEmitter;
    }

    public boolean sendMessage(String uid,String messageId,String message){
        if(StringUtils.isEmpty(message)){
            log.info("[{}]参数异常，msg为空",uid);
            return false;
        }
        SseEmitter sseEmitter = sseEmitterMap.get(uid);
        if(sseEmitter == null){
            log.info("[{}]sse连接不存在",uid);
            return  false;
        }
        try{
            sseEmitter.send(SseEmitter.event().id(messageId).reconnectTime(60000).data(message));
            log.info("用户{},消息ID：{}，推送成功：{}",uid,messageId,message);
            return true;
        }catch (IOException e){
            sseEmitterMap.remove(uid);
            log.info("用户{},消息ID：{}，消息推送失败：{}",uid,messageId,message);
            sseEmitter.complete();
            return false;
        }
    }

    public void closeSse(String uid){
        if(sseEmitterMap.containsKey(uid)){
            SseEmitter sseEmitter = sseEmitterMap.get(uid);
            sseEmitter.complete();
            sseEmitterMap.remove(uid);
        }else {
            log.info("用户{}连接已关闭",uid);
        }
    }

}

Controller:

package com.zillion.aggregate.app.controller;


import cn.hutool.core.util.IdUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

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

@Controller
@RequestMapping("/aggregate/api/pay")
public class TestController {

    private static final Map<String,Boolean> SEND_MAP = new ConcurrentHashMap<>();

    @Autowired
    private SSEService sseService;
    @GetMapping("createSse")
    @CrossOrigin
    public SseEmitter createSse(String uid)
    {
        return sseService.crateSse(uid);
    }

    @GetMapping("/sendMsg")
    @ResponseBody
    @CrossOrigin
    public SseEmitter sendMsg(@RequestParam("uid") String uid) throws InterruptedException {
        SseEmitter sseEmitter = sseService.crateSse(uid);
        if (SEND_MAP.get(uid)==null ||  !SEND_MAP.get(uid)){
             new Thread(()->{
                 int i=0;
                 while (true){
                     try {
                         i++;
                         String message = "uid:"+uid+" number:"+i+" message:"+IdUtil.fastUUID().replace("-", "");
                         sseService.sendMessage(uid,"消息"+i,message);
                         SEND_MAP.put(uid,true);
                         Thread.sleep(1000);
                     } catch (InterruptedException e) {
                         e.printStackTrace();
                         closeSse(uid);
                     }
                 }
             }).start();
         }

        return sseEmitter;
    }

    @GetMapping("closeSse")
    @CrossOrigin
    public void closeSse(String uid){
        sseService.closeSse(uid);
    }
}

5.前端实现消息监听

<!doctype html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1">
    <title>SSE消息推送监听</title>
</head>
<body>
    <div id="conMsg"></div>
<script>
    let uid = 1;
    let chat = document.getElementById("conMsg");
    if(window.EventSource){
        var eventSource = new EventSource(`http://localhost:9001/aggregate/aggregate/api/pay/sendMsg?interfaceId=CEDB297CECCC9DCBAD348204ACDD5BAD&uid=${uid}`);
        eventSource.onopen = ()=>{
            console.log("链接成功");
        }
        eventSource.onmessage = (ev)=>{
            if(ev.data){
                chat.innerHTML += ev.data+"<br>";
            }
        }
        eventSource.onerror = ()=>{
            console.log("sse链接失败")
        }
    }else{
        alert("当前浏览器不支持sse")
    }
</script>
</body>
</html>

目录

JAVA通过SSE实现消息推送

JAVA通过SSE实现消息推送

1.什么是SSE？

2.SSE技术的基本原理

3.SSE和Socket的区别

4.编写SSE服务，来进行创建链接和发送消息

5.前端实现消息监听