概述
1、封裝springboot全局的消息推送接口；
注：1）由于原生HTML5 EventSource 不支持添加header，所以要把連接創建接口加入身份驗證白名單，并在接口內添加自己校驗token2）后臺需定時心跳，保證鏈接的存活
2、封裝前端公共的消息推動存儲方法：保證整個系統只有1個消息鏈接
組件可根據傳遞指定的業務類型，展示制定的消息
3、注意sse連接建立接口，需要單獨指定nginx配置，防止nginx默認配置導致的推送鏈接中斷
4、分布式系統，該后臺接口改動介紹測試效果
如下圖
1 后端接口的實現
controller有3個方法
1、sse鏈接建立
2、給已連接的指定用戶推送消息（用戶在線才能收到，不在線消息丟下：可根據您的業務再做具體代碼編寫）
3、給所有已建立的用戶廣播消息
注：本文章采用：有心跳 → 用 0L 永久連接，服務器資源受控，客戶端也能保持連接
也可采用：無心跳 → 建議設置 30~60 秒超時，客戶端需要重連：適合連接數量非常多，或者不頻繁推送的場景
1.1 推送服務Service
SseService 接口
package com.server.common.notice.service;import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;public interface SseService {/*** 建立連接** @param clientId 連接id，這里是用戶id* @return see 服務器事件響應*/SseEmitter connect(String clientId);/*** 給指定 連接，發送消息** @param clientId 連接id* @param type 消息類型* @param data 數據*/void sendMessage(String clientId, String type, Object data);/*** 廣播消息** @param type 類型* @param data 數據*/void broadcast(String type, Object data);
}SseService 接口實現
注意：鏈接建立邏輯不要做改動，若直接根據clientId 移除和關閉，可能造成競態刪除”錯誤對象
package com.server.common.notice.service.impl;import com.server.common.notice.service.SseService;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;@Service
public class SseServiceImpl implements SseService {private final Map<String, SseEmitter> clients = new ConcurrentHashMap<>();@Overridepublic SseEmitter connect(String clientId) {// 1) 先移除舊連接，避免 onCompletion 把后續的新連接誤刪SseEmitter old = clients.remove(clientId);if (old != null) {try {old.complete();} catch (Exception ignore) {}}// 2) 建立新連接（可改成有超時，比如 15min；配合心跳更穩）SseEmitter emitter = new SseEmitter(0L);clients.put(clientId, emitter);// 3) 回調里做“條件刪除”：僅當 Map 中的值就是當前這個 emitter 時才刪除Runnable cleanup = () -> clients.remove(clientId, emitter);emitter.onCompletion(cleanup);emitter.onTimeout(cleanup);emitter.onError(ex -> cleanup.run());// 初始事件try {emitter.send(SseEmitter.event().name("INIT").data("connected"));} catch (Exception e) {try {emitter.completeWithError(e);} catch (Exception ignore) {}}return emitter;}@Overridepublic void sendMessage(String clientId, String type, Object data) {SseEmitter emitter = clients.get(clientId);if (emitter == null) return;try {emitter.send(SseEmitter.event().name("MESSAGE").data(Map.of("id", UUID.randomUUID().toString(),"type", type,"data", data)));} catch (Exception e) {clients.remove(clientId, emitter);try {emitter.completeWithError(e);} catch (Exception ignore) {}}}@Overridepublic void broadcast(String type, Object data) {for (String clientId : clients.keySet()) {sendMessage(clientId, type, data);}}
}
1.2 推送服務Controller
package com.server.common.notice.controller;import cn.hutool.core.util.ObjectUtil;
import com.server.common.notice.service.SseService;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;import java.time.Instant;@Slf4j
@RestController
@RequestMapping("/common/stream")
public class SseController {@Resourceprivate SseService sseService;/*** 建立 SSE 連接** @param clientId 連接id* @return sse 事件*/@GetMapping("/connect/{clientId}")public SseEmitter connect(@PathVariable String clientId, @RequestParam String token) {// todo 編寫自己的token校驗，且不加刷新token邏輯，否則系統永不掉線return sseService.connect(clientId);}/*** 給指定用戶推送（僅測試用）** @param clientId 連接id* @param type 類型* @param data 數據*/@PostMapping("/push/{clientId}")public void push(@PathVariable String clientId,@RequestParam String type,@RequestBody Object data) {sseService.sendMessage(clientId, type, data);}/*** 廣播推送（僅測試用）** @param type 類型* @param data 數據*/@PostMapping("/broadcast")public void broadcast(@RequestParam String type,@RequestBody Object data) {sseService.broadcast(type, data);}
}
1.3 定時心跳，保證鏈接不中斷
package com.server.common.notice.schedule;import com.server.common.notice.service.SseService;
import jakarta.annotation.Resource;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;@Component
public class SseHeartbeatTask {@Resourceprivate SseService sseService;/*** 每30秒執行一次，給sse鏈接事件，發送一次廣播*/@Scheduled(fixedRate = 30000)public void sendHeartbeat() {sseService.broadcast("HEARTBEAT", "ping");}
}
2 前端公共組件封裝
核心：
1、使用公共變量存儲前端唯一的：eventSource ，不重復建立連接
2、pinia 類定義了前端唯一的sse事件監聽器，且存儲所有消息
3、封裝消息展示組件固定化流程，防止鏈接重復創建
包含：sseStore.js、PushMessage.vue
2.1 sseStore.js
注：其中鏈接地址的前綴需要你根據實際業務調整：例如：import.meta.env.VITE_BASE_API
屬性 messages：存儲的消息列表：格式：[{id:唯一標識，type:消息類型,data:消息內容}]
屬性 connected：是夠鏈接建立成功
方法：
1）initSse 初始化鏈接，傳遞用戶id 和 用戶的token
2）removeMessage:存儲中刪除指定消息
import {defineStore} from 'pinia';
import {ref} from 'vue';let eventSource = null; // 全局唯一 SSE 連接export const useSseStore = defineStore('sse', () => {const messages = ref([]);const connected = ref(false);function initSse(clientId, token) {if (eventSource) return; // 已存在連接eventSource = new EventSource(`/api/common/stream/connect/${clientId}?token=${token}`);eventSource.addEventListener('INIT', () => {connected.value = true;console.log('SSE connected');});eventSource.addEventListener('MESSAGE', event => {const msg = JSON.parse(event.data);messages.value.push(msg);if (messages.value.length > 500) messages.value.shift(); // 限制緩存});eventSource.addEventListener('HEARTBEAT', () => {// 可更新心跳狀態});eventSource.onerror = () => {connected.value = false;console.warn('SSE error, will auto-reconnect');};}function removeMessage(id) {messages.value = messages.value.filter(msg => msg.id !== id);}return {messages, connected, initSse, removeMessage};
});
2.2 PushMessage.vue
注：onMounted的參數需要你根據實際業務傳遞，ui展示也需要您根據業務調整
<template><div class="push-messages"><h4>{{ title }},鏈接狀態：{{ sseStore.connected }}</h4><ul><li v-for="msg in filteredMessages" :key="msg.id">{{ msg.data }}<button @click="remove(msg.id)">刪除</button></li></ul><p v-if="!filteredMessages.length">暫無消息</p></div>
</template><script setup>
import {computed, onMounted} from 'vue';
import {useSseStore} from '@/pinia/sseStore.js';const props = defineProps({// 消息類型moduleType: {type: String, required: true},// 組件標題title: {type: String, default: '消息推送'},// 最大緩存數量maxCache: {type: Number, default: 50}
});const sseStore = useSseStore();// 過濾指定模塊消息
const filteredMessages = computed(() => {return sseStore.messages.filter(msg => msg.type === props.moduleType).slice(-props.maxCache);
});function remove(id) {sseStore.removeMessage(id);
}// 組件掛載時調用 initSse
onMounted(() => {// todo 這里需要根據你的實際情況：傳遞用戶id 和需要校驗的tokensseStore.initSse()
})
</script><style scoped>
.push-messages {border: 1px solid #ddd;padding: 10px;max-height: 300px;overflow-y: auto;
}.push-messages ul {list-style: none;padding: 0;margin: 0;
}.push-messages li {display: flex;justify-content: space-between;padding: 4px 0;border-bottom: 1px dashed #eee;
}.push-messages button {background: #f5f5f5;border: 1px solid #ccc;padding: 2px 5px;cursor: pointer;
}
</style>
3 前端組件測試
包含：1 sseApi.js 本質就是網路請求公共提取
2 SseMessageTest.vue，測試頁面
3.1 sseApi.js
這里引用的utils/request會自動添加header 頭
import request from '@/utils/request'export default {// 給用發送消息sendToUser(clientId, type, data) {return request({url: `/common/stream/push/${clientId}?type=${type}`,method: 'post',data: data,headers: {'Content-Type': 'application/json', // 根據需求設置}})},//broadcast(type, data) {return request({url: `/common/stream/broadcast?type=${type}`,method: 'post',data: data,headers: {'Content-Type': 'application/json', // 根據需求設置}})},
}
3.1 SseMessageTest.vue
下列：common.getUserIdByToken() 為我這獲取當前登陸用戶id的前端方法，請根據實際業務進行替換
<template><el-card class="sse-message-test" shadow="hover"><template #header><span>📡 SSE 消息測試，用戶：{{ common.getUserIdByToken() }}</span></template><el-divider content-position="left">1 給指定用戶發消息</el-divider><!-- 給指定用戶發消息 --><el-form :inline="true" :model="formSingle" class="form-block"><el-form-item label="用戶ID"><el-input v-model="formSingle.clientId" placeholder="請輸入用戶ID" style="width: 200px"/></el-form-item><el-form-item label="類型"><el-input v-model="formSingle.type" placeholder="如 chat/order" style="width: 160px"/></el-form-item><el-form-item label="消息"><el-input v-model="formSingle.data" placeholder="請輸入消息內容" style="width: 260px"/></el-form-item><el-form-item><el-button type="primary" @click="sendToUser">發送給用戶</el-button></el-form-item></el-form><el-divider/><el-divider content-position="left">2 給指所有用戶廣播消息</el-divider><!-- 廣播消息 --><el-form :inline="true" :model="formBroadcast" class="form-block"><el-form-item label="類型"><el-input v-model="formBroadcast.type" placeholder="如 notice/chat" style="width: 160px"/></el-form-item><el-form-item label="消息"><el-input v-model="formBroadcast.data" placeholder="請輸入廣播內容" style="width: 260px"/></el-form-item><el-form-item><el-button type="success" @click="broadcast">廣播所有人</el-button></el-form-item></el-form><el-divider content-position="left">3 收到的指定消息</el-divider><push-message module-type="chat"/><el-divider content-position="left">4 收到的廣播消息</el-divider><el-divider content="廣播信息"/><push-message module-type="notice"/></el-card>
</template><script setup>
import {reactive} from 'vue'
import {ElMessage} from 'element-plus'
import sseApi from '@/api/sys/sseApi.js'
import common from "@/utils/common.js";import PushMessage from "@/components/message/PushMessage.vue";// 單用戶消息
const formSingle = reactive({clientId: common.getUserIdByToken(),type: 'chat',data: ''
})// 廣播消息
const formBroadcast = reactive({type: 'notice',data: ''
})// 給指定用戶發消息
async function sendToUser() {if (!formSingle.clientId || !formSingle.data) {return ElMessage.warning('請填寫用戶ID和消息內容')}try {await sseApi.sendToUser(formSingle.clientId, formSingle.type, formSingle.data)ElMessage.success(`已向 ${formSingle.clientId} 發送消息`)} catch (e) {ElMessage.error('發送失敗')}
}// 廣播所有人
async function broadcast() {if (!formBroadcast.data) {return ElMessage.warning('請輸入廣播內容')}try {await sseApi.broadcast(formBroadcast.type, formBroadcast.data)ElMessage.success('廣播成功')} catch (e) {console.log('廣播失敗', e)ElMessage.error('廣播失敗：')}
}
</script><style scoped>
.sse-message-test {margin: 20px;
}.form-block {margin-bottom: 15px;
}
</style>
3 nginx部署改動
單獨添加
# SSE 專用location /api/common/stream/connect/ {proxy_set_header Host $http_host;proxy_set_header X-Real-IP $remote_addr;proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;# SSE 專用配置proxy_http_version 1.1; # SSE 必須 HTTP/1.1proxy_set_header Connection ''; # 保持長連接chunked_transfer_encoding off; # 保證消息實時# 本機上運行的后端接口proxy_pass http://127.0.0.1:8080/common/stream/connect/;}
完整的配置nginx的配置，這邊nginx安裝是啟用了壓縮
詳見：https://blog.csdn.net/qq_26408545/article/details/133685624?spm=1001.2014.3001.5502
# Nginx 進程數，一般設置為和 CPU 核數一樣，可設置 auto
worker_processes auto;#error_log logs/error.log; # Nginx 的錯誤日志存放目錄
#error_log logs/error.log notice;
#error_log logs/error.log info;#pid logs/nginx.pid; # Nginx 服務啟動時的 pid 存放位置events {# 根據操作系統自動選擇：建議指定事件驅動模型，避免 Nginx 誤判環境use epoll;# 每個進程允許最大并發數# 小規模的服務器：512或1024，中等規模的服務器：2048或4096，大規模的服務器：8192或更高# 考慮到內存占用和CPU的利用率，一般建議不要將worker_connections設置得過高worker_connections 2048;# 默認：off，高并發下建議開，讓 worker 每次盡量多 accept 新連接multi_accept on;# 默認：on,避免多個 worker 同時搶占 accept，減少驚群現象accept_mutex on;
}http {include mime.types;# 文件擴展名與類型映射表default_type application/octet-stream;# 默認文件類型# 設置日志模式#log_format main '$remote_addr - $remote_user [$time_local] "$request" '# '$status $body_bytes_sent "$http_referer" '# '"$http_user_agent" "$http_x_forwarded_for"';#access_log logs/access.log main; # Nginx訪問日志存放位置sendfile on;# 開啟高效傳輸模式#tcp_nopush on;# 減少網絡報文段的數量keepalive_timeout 65;# 保持連接的時間，也叫超時時間，單位秒gzip on;#表示開啟壓縮功能gzip_static on;#靜態文件壓縮開啟# 設置壓縮的最低文件大小（默認值是 20 字節）gzip_min_length 5k;# 設置為 1KB 或更大，避免對小文件壓縮# 設置使用的壓縮算法（一般是 gzip）gzip_comp_level 7;# 范圍是 1-9，數字越大壓縮率越高，但占用 CPU 更多# 開啟對特定文件類型的壓縮(不建議壓縮緊湊格式：圖片)gzip_types text/plain text/css application/javascript application/json application/xml text/xml application/xml+rss text/javascript application/font-woff2 application/font-woff application/font-otf;# 不壓縮的 MIME 類型gzip_disable "msie6";# 禁止壓縮 IE6 瀏覽器# 壓縮緩存控制gzip_vary on;# 設置響應頭 `Vary: Accept-Encoding`# 壓縮后文件傳輸gzip_buffers 16 8k;# 設定緩沖區大小#認證后臺server {listen 80; # 88 ssl 本服務監聽的端口號server_name localhost; # 主機名稱client_max_body_size 600m;client_body_buffer_size 128k;proxy_connect_timeout 600;proxy_read_timeout 600;proxy_send_timeout 600;proxy_buffer_size 64k;proxy_buffers 4 32k;proxy_busy_buffers_size 64k;proxy_temp_file_write_size 64k;# 首頁 index.html — 禁止緩存，強烈推薦location = /index.html {root /opt/sm-crypto/process-center-web/dist;add_header Cache-Control "no-cache, no-store, must-revalidate";add_header Pragma "no-cache";add_header Expires "0";try_files $uri =404;}# 靜態資源 /assets/，緩存7天，不帶immutable，允許刷新更新location /assets/ {root /opt/sm-crypto/process-center-web/dist;expires 7d;add_header Cache-Control "public";}location / {# root 規定了通過監聽的端口號訪問的文件目錄root /opt/sm-crypto/process-center-web/dist;# 配置資源重新跳轉，防止刷新后頁面丟失try_files $uri $uri/ /index.html;# index 規定了該目錄下指定哪個文件index index.html index.htm;}# SSE 專用location /api/common/stream/connect/ {proxy_set_header Host $http_host;proxy_set_header X-Real-IP $remote_addr;proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;# proxy_http_version 1.1; # SSE 必須 HTTP/1.1proxy_set_header Connection ''; # 保持長連接chunked_transfer_encoding off; # 保證消息實時# 本機上運行的后端接口proxy_pass http://127.0.0.1:8080/common/stream/connect/;}# 配置后端接口的跨域代理# 對于路徑為 "api 的接口，幫助他跳轉到指定的地址location /api/ {proxy_set_header Host $http_host;proxy_set_header X-Real-IP $remote_addr;proxy_set_header REMOTE-HOST $remote_addr;proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;# 本機上運行的后端接口proxy_pass http://127.0.0.1:8080/; }location /status{stub_status on;}}}
4 分布式系統，該后臺接口改動介紹
實現有多重可問AI，下面只是實現方案之一
基于 消息隊列（推薦）
- 核心思路
- 每臺應用實例只維護自己的 SSE 連接
- 推送消息通過 消息中間件（Redis Pub/Sub、Kafka、RabbitMQ 等）廣播到所有節點
- 每臺節點收到消息后，將其推送給本節點內存里的 SSE 客戶端
- 流程示意
客戶端(EventSource)|v節點 A --------> Redis Pub/Sub ---------> 節點 B| |v vSseEmitter SseEmitter
- 優點
- 高可用，自動擴展節點
- 節點之間解耦
- 消息順序可控（Kafka 支持順序）
- 實現舉例（Redis Pub/Sub）
// 發布消息
redisTemplate.convertAndSend("sse:channel", msg);// 訂閱消息
@EventListener
public void onMessage(Message msg) {sseService.broadcast(msg.getType(), msg.getData());
}
