SSE(Server-Sent Events)复用EventSource方法及代码优化咨询

一、EventSource跨标签页复用实现方案 #

浏览器不同标签页属于独立的运行上下文，默认无法直接共享EventSource实例，可通过以下两种方案实现复用：

• 方案1：使用Shared Worker（推荐）
  Shared Worker是浏览器原生支持的多标签页共享后台线程，你可以在Shared Worker内部仅创建1个EventSource实例监听后端推送，所有打开的标签页都和该Worker建立通信，接收Worker转发的SSE消息即可，全程仅维持1个SSE连接。
• 方案2：主标签页选举 + Broadcast Channel
  所有标签页通过Broadcast Channel通信，共同选举一个存活的标签页作为「主节点」，仅主节点创建EventSource，收到推送后通过Broadcast Channel广播给其余所有标签页。主标签页关闭时其余标签页自动重新选举新的主节点即可。

二、现有代码优化建议 #

前端代码优化 #

原代码存在以下问题：

1. EventSource实例没有持久化存储，组件卸载时无法主动关闭，会造成内存泄漏
2. 没有添加错误监听、断开重连逻辑
3. 未处理SSE连接关闭场景
   优化后代码如下：

import axios from 'axios'
import React, { useEffect, useState, useRef, useCallback } from 'react'

const FrontendTest = () => {
  const [data, setData] = useState(null)
  const [databaseInfo, setDatabaseInfo] = useState(null)
  const eventSourceRef = useRef(null)
  const SSE_URL = 'http://localhost:5000/test'

  // 初始化SSE连接
  const initSSE = useCallback(() => {
    if (eventSourceRef.current?.readyState === EventSource.OPEN) return
    const source = new EventSource(SSE_URL)
    source.onmessage = (e) => {
      setData(JSON.parse(e.data))
    }
    source.onerror = (err) => {
      console.error('SSE连接出错，5秒后重连:', err)
      source.close()
      setTimeout(initSSE, 5000)
    }
    eventSourceRef.current = source
  }, [])

  useEffect(() => {
    initSSE()
    // 组件卸载时关闭SSE连接
    return () => {
      eventSourceRef.current?.close()
    }
  }, [initSSE])

  const buttonClicked = async (e) => {
    e.preventDefault()
    const number = Math.random()
    const sendReq = await axios.post('http://localhost:5000/savedata', { number })
    setDatabaseInfo(sendReq.data)
  }

  return (
    <div>
      <div>
        <button onClick={buttonClicked}>Send</button>
        <p>如果数值大于0.5会被筛选推送</p>
        <p>
          {databaseInfo && `${databaseInfo.data.toFixed(4)} 已保存到数据库!`}
        </p>
        <p>
          {data && `${data.toFixed(4)} 已匹配到! 对应数据库记录已删除!`}
        </p>
      </div>
    </div>
  )
}

export default FrontendTest

后端代码优化 #

原代码存在严重的逻辑缺陷：

1. 全局interval变量会被新的SSE请求覆盖，导致只有最后一个连接的客户端能收到推送，之前的连接会全部失效
2. 没有处理客户端断开连接的场景，interval不会被清除，造成内存泄漏
3. 每3秒全表扫描数据库性能极低，实时性差
4. 缺少SSE必要的响应头配置
   优化后代码如下：

import express from 'express'
import cors from 'cors'
import expressAsyncHandler from 'express-async-handler'
import mongoose from 'mongoose'
import Datas from './model.js'

const app = express()
const port = 5000
const CONNECTION_URL = '你的MongoDB连接地址'

app.use(
  cors({
    origin: 'http://localhost:3000',
    credentials: true,
  })
)
app.use(express.json())
app.use(express.urlencoded({ extended: true }))

app.post(
  '/savedata',
  expressAsyncHandler(async (req, res) => {
    const data = req.body.number
    await Datas.create({ data1: data })
    res.send({ data })
  })
)

app.get(
  '/test',
  expressAsyncHandler(async (req, res) => {
    // 设置SSE必要响应头
    res.setHeader('Content-Type', 'text/event-stream')
    res.setHeader('Cache-Control', 'no-cache')
    res.setHeader('Connection', 'keep-alive')
    res.setHeader('Access-Control-Allow-Origin', 'http://localhost:3000')

    // 每个连接维护独立的定时任务
    const interval = setInterval(async () => {
      // 直接查询大于0.5的记录，不需要全表扫描
      const targetRecords = await Datas.find({ data1: { $gt: 0.5 } })
      for (const record of targetRecords) {
        res.write(`data: ${JSON.stringify(record.data1)}\n\n`)
        await Datas.findByIdAndDelete(record._id)
      }
      console.log('扫描完成')
    }, 3000)

    // 监听客户端断开连接，清除定时任务
    req.on('close', () => {
      clearInterval(interval)
      res.end()
    })
  })
)

mongoose
  .connect(CONNECTION_URL, { useNewUrlParser: true, useUnifiedTopology: true })
  .then(() =>
    app.listen(port, () =>
      console.log(`服务运行在 http://localhost:${port}`)
    )
  )
  .catch((error) => console.log(error))

如果想要进一步提升性能，可以使用MongoDB的ChangeStream监听数据插入事件，不需要定时轮询，实时性更高，也不会产生无效扫描。 #

内容的提问来源于stack exchange，提问作者caner küncü