一、后端
1、新建工具类SseServiceTool

package com.example.system_manage.utils;import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;/*** SSE消息推送工具*/
@Slf4j
@Component
public class SseServiceTool {private final static Map<String, SseEmitter> sseEmitterMap = new ConcurrentHashMap<>();//创建一个SSE对象public SseEmitter createSse(String messageId) {closeSseEmitterById(messageId);log.info("创建对象");log.info(sseEmitterMap.size()+"");SseEmitter sseEmitter = new SseEmitter(5_60_000L);// 设置前端的重试时间为1ssseEmitter.onCompletion(()->{log.info("完成传输");});sseEmitter.onError((throwable)->{log.info("出现错误");});sseEmitterMap.put(messageId, sseEmitter);return sseEmitter;}public SseEmitter getSseEmitterById(String id){if(!sseEmitterMap.containsKey(id)){throw new BusinessException("当前无连接对象");}return sseEmitterMap.get(id);}//根据id关闭连接public void closeSseEmitterById(String messageId){if(sseEmitterMap.containsKey(messageId)){sseEmitterMap.get(messageId).complete();sseEmitterMap.remove(messageId);}}
}

2、新建service接口

package com.example.system_manage.service;import java.util.Map;public interface ISseMessageService {//发送消息String pushMessage(Map<String,String> map);//关闭SSE服务String closeSseService(String messageId);
}

3、新建service实现类

package com.example.system_manage.service.impl;import com.example.system_manage.service.ISseMessageService;
import com.example.system_manage.utils.SseServiceTool;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;import java.util.Map;@Service
@Slf4j
public class SseMessageServiceImpl implements ISseMessageService {@Resourceprivate SseServiceTool sseServiceTool;@Overridepublic String pushMessage(Map<String, String> map) {try {SseEmitter sseEmitter = sseServiceTool.getSseEmitterById(map.get("id"));Map<String,String> sendMessage=Map.of("title",map.get("title"),"content",map.get("content"));SseEmitter.SseEventBuilder data = SseEmitter.event().name(map.get("eventName")).id(map.get("id")).data(sendMessage);sseEmitter.send(data);}catch (Exception ex){//移除连接sseServiceTool.closeSseEmitterById(map.get("id"));log.error(ex.getMessage());return "noSSE";}return "OK";}@Overridepublic String closeSseService(String messageId) {sseServiceTool.closeSseEmitterById(messageId);return "OK";}
}

4、新建控制器

package com.example.system_manage.controller;import com.example.system_manage.service.ISseMessageService;
import com.example.system_manage.utils.ResultMap;
import com.example.system_manage.utils.SseServiceTool;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;import java.util.List;
import java.util.Map;@RestController
@RequestMapping(path = "/sseMessage")
@Slf4j
public class SseMessageController {@Resourceprivate ISseMessageService sseMessageService;@Resourceprivate SseServiceTool sseServiceTool;//创建SSE连接对象@GetMapping(path = "/createSse", produces = {MediaType.TEXT_EVENT_STREAM_VALUE})public SseEmitter createSse(@RequestParam("messageId") String messageId) {return sseServiceTool.createSse(messageId);}//推送消息@PostMapping(path = "/pushMessage")public ResultMap pushMessage(@RequestBody Map<String,String> map) {return ResultMap.SUCCESS.setNewData(sseMessageService.pushMessage(map));}//关闭SSE连接@GetMapping("/closeSseService")public ResultMap closeSseService(@RequestParam("messageId") String messageId){return ResultMap.SUCCESS.setNewData(sseMessageService.closeSseService(messageId));}}

二、前端部分
1、安装依赖

 npm install event-source-polyfill  -S

2、封装requestEventSource.js用于创建SSE对象

import { EventSourcePolyfill } from 'event-source-polyfill';
import cookie from 'js-cookie';/*** 创建一个sse对象* @param {*} data* @returns*/
export const getEventSource = (data) => {return new EventSourcePolyfill(`${data.httpRequest}${data.url}`, {headers: {'xxx': cookie.get("xxxxx"),},});
};

用于获取请求SSE对象

export function getSseMessageObj(data) {return getEventSource({httpRequest: baseUrl,url: `/sseMessage/createSse?messageId=${data}`,});
}

3、页面组件

<!--* @Author: zhangming zhangming@sinoma-tianjin.cn* @Date: 2024-07-15 15:48:08* @LastEditors: zhangming zhangming@sinoma-tianjin.cn* @LastEditTime: 2024-07-16 15:35:23* @FilePath: \zjcgg_system_manage\src\views\demos\OrganizationSelectTest.vue* @Description: 这是默认设置,请设置`customMade`, 打开koroFileHeader查看配置 进行设置: https://github.com/OBKoro1/koro1FileHeader/wiki/%E9%85%8D%E7%BD%AE
-->
<template>SSE测试<el-button type="primary" @click="pushSseMessageTopLeft">左上角推送消息</el-button><el-button type="primary" @click="pushSseMessageTopRight">右上角推送消息</el-button>
</template><script setup>
import { ref, onMounted, onBeforeMount, onBeforeUnmount, nextTick, reactive, watch } from 'vue';
import { EventSourcePolyfill } from 'event-source-polyfill';
import { ElMessage, ElNotification } from 'element-plus';
import { generateUUID } from '@/utils/index';
import { closeSseService, pushSseMessage, getSseMessageObj } from '@/api/sseClient';const source = ref(null); //sse连接对象
const messageKey = ref('');window.addEventListener('beforeunload', async (event) => {//浏览器窗口事件// 设置returnValue属性可以显示一个提示信息，询问用户是否真的要离开页面await closeSsse();
});window.addEventListener('unload', async (event) => {//浏览器窗口事件// 设置returnValue属性可以显示一个提示信息，询问用户是否真的要离开页面await closeSsse();
});onBeforeMount(async () => {//初始加载messageKey.value = generateUUID();source.value = getSseMessageObj(messageKey.value);source.value.addEventListener('message-top-left',(event) => {let response = JSON.parse(event.data);ElNotification({title: response.title,message: response.content,position: 'top-left',});},false);source.value.addEventListener('message-top-right',(event) => {let response = JSON.parse(event.data);ElNotification({title: response.title,message: response.content,position: 'top-right',});},false);
});onBeforeUnmount(async () => {await closeSsse();
});const closeSsse = async () => {//获取到选择的人source.value.close();await closeSseService(messageKey.value);
};const pushSseMessageTopRight = async () => {let postData = {id: messageKey.value,content: '报警通知',title: '消息主题',eventName: 'message-top-right',};await pushSseMessage(postData);
};const pushSseMessageTopLeft = async () => {let postData = {id: messageKey.value,content: '报警通知',title: '消息主题',eventName: 'message-top-left',};await pushSseMessage(postData);
};
</script><style lang="scss" scoped></style>

注：里面有部分接口是请求后端断开连接用的