思路
一个基于订阅发布机制的SSE事件。客户端可以请求订阅api(携带客户端id),与服务器建立SSE链接;后续服务器需要推送消息到客户端时,再根据客户端id从已建立链接的会话中找到目标客户端,将消息推送出去。
后端
这个控制器类允许客户端订阅、接收消息和断开连接,通过 pool
存储 SseEmitter
并对其进行管理。
package com.example.q11e.controller;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import java.io.IOException;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
@RestController
public class SseController {
// 存储已订阅的客户端的会话列表
private final Map<String, SseEmitter> pool = new ConcurrentHashMap<>();
// 向特定的 SseEmitter 发送消息
public void publisher(String id_sid, int content) {
// 根据 id_sid 从映射中获取 SseEmitter
SseEmitter sseEmitter = pool.get(id_sid);
if (Objects.isNull(sseEmitter)) {
return;
}
try {
sseEmitter.send(content); // 发送内容
} catch (IOException e) {
System.out.println("null " + e);
}
}
// 处理客户端的订阅请求
@GetMapping("/subscribe/{id}")
public SseEmitter subscribe(@PathVariable("id") String id_sid) {
// 根据 id_sid 从映射中获取 SseEmitter
SseEmitter sseEmitter = pool.get(id_sid);
if (Objects.isNull(sseEmitter)) {
// 如果不存在,则创建一个新的 SseEmitter,设置超时时间为 130000 毫秒
sseEmitter = new SseEmitter(130000L);
// 设置发送完成事件:从映射中移除该 SseEmitter
sseEmitter.onCompletion(() -> pool.remove(id_sid));
// 设置超时事件:从映射中移除该 SseEmitter
sseEmitter.onTimeout((() -> pool.remove(id_sid)));
// 将新创建的 SseEmitter 放入映射中
pool.put(id_sid, sseEmitter);
}
// System.out.println(pool);
// 返回 SseEmitter 给客户端
return sseEmitter;
}
// 处理客户端的断开连接请求
public void disconnect(String id_sid) {
SseEmitter emitter = pool.remove(id_sid);
if (emitter!= null) {
emitter.complete();
}
}
}
发送消息
package com.example.q11e.service;
import com.example.q11e.controller.SseController;
@Service
public class BatchService {
@Autowired
public BatchService(SseController sseController) {
this.sseController = sseController;
}
private final SseController sseController;
@Async
public void batchRequests(){
// uid+"_"+sid 客户端标识符,sucCount为需要发送的信息
sseController.publisher(uid + "_" + sid, sucCount);
sseController.disconnect(uid + "_" + sid);
}
}
前端
SSE状态管理 store.ts
// sse前端
import { defineStore } from 'pinia';
import { getUserBalance } from '@/request/api.ts'
export const useESStore = defineStore('EventSource', {
state: () => ({
uid: localStorage.getItem('uid'),
balance: 1,
eventSourceInstance: null as EventSource | null, // 新增状态属性
currentSid: null as string | null,
currentCount: 0,
currentTotal: 0,
connect: false
}),
actions: {
setUid(uid:string) {
this.uid = uid;
},
setConnect(connect: boolean) {
this.connect = connect
},
initEventSource(sid:string) {
if (this.uid) {
const sseURL = import.meta.env.VITE_SSE_URL
const evtSrcInstance = new EventSource(sseURL + "/" + this.uid + "_" + sid);
evtSrcInstance.onmessage = (e) => {
this.setCurrentCount(e.data) //普通函数时: this-->evtSrcInstance
};
evtSrcInstance.onopen = () => {
this.setCurrentCount(0)
this.setConnect(true)
};
evtSrcInstance.onerror = () => {
this.setConnect(false)
this.setCurrentTotal(0)
};
this.eventSourceInstance = evtSrcInstance; // 存储实例到状态
}
},
closeEventSource() {
if (this.eventSourceInstance) {
this.eventSourceInstance.close();
this.eventSourceInstance = null;
}
}
}
});
<template>
<span v-show="connect">
<span class="sid">{{ sid }}</span>
<span v-for="(char, index) in ['.', '.', '.']" :key="index" class="blink-effect sid"
:style="{ animationDelay: `${index * 0.1}s` }">{{ char }}</span>
<!----count是服务器推送的内容----->
<span class="process">{{ count }}/{{ total }}</span>
</span>
</template>
<script lang="ts" setup>
import { computed } from 'vue'
import { useESStore } from '@/store/store.ts'
const SSE = useESStore()
let count = computed(() => SSE.currentCount)
let total = computed(() => SSE.currentTotal)
let sid = computed(() => SSE.currentSid)
let connect = computed(() => SSE.connect)
</script>
<style scoped>
.process {
background: red;
color: white;
padding: 2px 4px;
}
.sid { color: #333; }
@keyframes blink {
0%, 100% {
transform: translateY(0); /* 开始和结束状态位置无变化 */
}
50% {
transform: translateY(-5px); /* 中间状态位置向上移动5px */
}
}
.blink-effect {
display: inline-block;
animation: blink 1s infinite;
}
</style>
效果
后端执行某耗时任务时,需要实时推送进度到客户端,每完成一个阶段,就向客户端推送一个单位进度,做到客户端实时显示进度的效果。