企业官网建设流程全解析

当我们开ai agent的时候需要从后端推送给前端消息，通过sse ,如果后端项目是运行的多进程，此时对于多进程下的sse连接管理，和前端连接，就需要一个完整的方案，如果处理不好就会出现冗余连接未删除，后端通知丢失，前端频繁触发后端close等问题，本文则是针对这些实际问题总结出解决方法

1.前端连接频繁触发后端sse close事件

需要设置openWhenHidden: true前端fetchEventSource 选项 防止页面切换 隐藏 会断联 从而触发了abort

fetchEventSource(import.meta.env.VITE_APP_BASEURL+'/sseConnect', { method: 'GET', headers: { 'Authorization': `Bearer ${token}`, 'Accept': 'text/event-stream' }, openWhenHidden: true,// 当页面隐藏或切换时保持连接状态 signal: currentController.signal,

前端对sse onerror的正确处理 防止报错和重连交叉使用导致的死循环触发重连,直接return不会触发前端的自动重连

fetchEventSource(import.meta.env.VITE_APP_BASEURL+'/sseConnect', { .... onerror(err) { console.error('SSE 连接错误:errrrrr', err); // 停止自动重连（关键！）再error里直接return 配合手动停止 abort 方法 不然会触发自动重连 // 当前端手动关闭时 不要进行重试 if( err.name === 'AbortError' ||currentController.signal.aborted){ console.log('连接被主动中断'); return } throw new Error('连接被主动中断'); // 可以根据需要提示用户连接失败 },

2. 关于多进程的sse 连接管理

进程间垃圾sse实例处理

每次重新连接调度到新的worker进程时 就删除原始进程上残留的当前用户的旧连接，

通过redis publish（[channelname],参数）广播 redis.subscribe（[channelname]） redis.on('message')核心通过 if (c && c.id !== exceptConnId){ 这行代码判断是删除的非当前worker进程上的残留僵尸sse实例 userid相同 但是exceptConnId 不同

const sseServers = new Map(); app.get('/sseConnect', async function (req, res, next) { let sseServer = new sseClient(res, req); pubClient = app.locals.redisClient; // 每次建立新连接都删除旧连接 let client = sseServers.get(req.signedCookies.sessionid); if (client) { try { client.res.end(); } catch {} sseServers.delete(req.signedCookies.sessionid); } // 通过sessionid 使用一个实例 sseServers.set(req.signedCookies.sessionid, sseServer); // 每次连接通知其他worker进程删除当前残留的连接 pubClient.publish('sse:kick', JSON.stringify({ userId: req.signedCookies.sessionid, exceptConnId: sseServer.id })); if (!subClient) { subClient = app.locals.redisClient.duplicate(); await subClient.subscribe('sse:kick'); subClient.on('message', async (channel, message) => { console.log('message:', message); if (channel == 'sse:kick') { // 从其他worker节点删除残留当前的sse连接 console.log('sse kick', message); const { userId, exceptConnId } = JSON.parse(message); const c = sseServers.get(userId); if (c && c.id !== exceptConnId) { try { c.res.end(); } catch {} sseServers.delete(userId); } }

命中通知活跃的sse

每次通知的时候进行所有进程的广播 命中到当前进程上存在的sse 进行连接通知，即当触发到业务sse通知逻辑后 每个worker节点都会走进监听回调，核心逻辑是判断是否命中的是当前worker进程上的sse 实例核心代码：

let client = sseServers.get(userId);
console.log('当前sse列表：', sseServers, '当前sse连接：', client);
// 本 worker 持有这个用户的 SSE 连接 → 推送
if (client) {

每一个进程都有一个sseServers队列，通过userId获取判断是否存在对应的userId sse实例 ，能找到唯一的sse ,因为结合上面的逻辑 已经把其他进程的相同userId的垃圾 sse实例都删除了

await subClient.subscribe(channelName); subClient.on('message', async (channel, message) => { if (channel == 'sse:broadcast') { console.log('sse redis', message); let payload; try { payload = JSON.parse(message); } catch (error) { console.log('sse redis error:', error); return; } let { userId, event, data } = payload; let client = sseServers.get(userId); console.log('当前sse列表：', sseServers, '当前sse连接：', client); // 本 worker 持有这个用户的 SSE 连接 → 推送 if (client) { try { // client.res.write(`event: ${event}\n`); // client.res.write(`data: ${JSON.stringify(data)}\n\n`); if (event == 'toolCall') { console.log('触发前端sse通知', JSON.stringify(data)); client.send('toolCall', JSON.stringify(data)); } } catch (err) { // 写失败说明连接已断，清理本地记录 sseServers.delete(userId); } } }

监听前端关闭后删除sse连接实例

监听前端关闭的时候删除连接删除的应该是当此创建的连接 防止close 触发延迟 即从当次创建A轮 到A轮的close期间有可能还会有B C创建 防止删除新的覆盖的连接
所以删除需要在onclose的时候判断sseServer=== sseClients.get(userId) 是否删除的是当轮次即A轮 创建的sse

app.get('/sseConnect', async function (req, res, next) { let sseServer = new sseClient(res, req); .... res.on('close', () => { // console.log('前端关闭sse请求！'); clearInterval(heartbeat); console.log(`[SSE close] userId=${req.signedCookies.sessionid} time=${Date.now()} writableEnded=${res.writableEnded} destroyed=${res.destroyed}`); console.log('前端关闭sse请求', sseServers, sseServers.get(req.signedCookies.sessionid)); // 删除旧的sse实例 必须删除！ // 否则再因为自动断连再重启后会因为找不到前端连接的对应的sse实例 // 判断获取的连接是否为当前的sse实例 防止删除新连接 const old = sseServers.get(req.signedCookies.sessionid); if (old === sseServer) { sseServers.delete(req.signedCookies.sessionid); } taskQuetMaps.delete(req.signedCookies.sessionid); });