律师事务所网站建设方案江门网站推广软件
业务需求
最近遇到了一个业务场景：用户在前端提交了一个耗时长的任务，后端将其转为后台任务并直接返回响应而不是等任务结束再返回响应。但是这样有个问题就是用户不知道什么时候任务完成了。根据这个场景，很容易就想到使用 SSE 来实现通知功能。
我的前后端技术栈分别是 Vue3 和 Python FastAPI。
什么是 SSE（Server-Sent Events）
SSE（Server-Sent Events） 是一种浏览器原生支持的、基于 HTTP 协议的 单向实时通信技术，允许服务器持续不断地向客户端推送数据。
它的典型使用场景包括：
- 实时通知（例如新消息提示、系统事件）
- 实时日志流（如部署日志）
- IoT 设备数据传输
与 WebSocket 相比，由于是单向通信（服务器到客户端）且浏览器原生支持，所以在实现推送消息时更加高效简单。
SSE 数据格式（EventStream）
SSE 推送数据是纯文本格式，遵循如下结构：
data: {"message": "Hello World"}
event: message
id: 12345
- data: 具体内容（可以是纯文本或 JSON 字符串）
- event: 事件类型（可选）
- id: 事件 ID，用于客户端重连时断点续传（可选）
SSE 和流式响应的概念区别
流式响应（Streaming Response）
是一种泛指，表示服务器不一次性返回完整响应，而是分段（chunked）返回响应体数据的方式。
常用于：大文件下载、大型计算任务的进度推送、LLM 流式输出
底层依赖：HTTP 的 Transfer-Encoding: chunked 或 HTTP/2 的流式机制
数据格式：不限，文本、JSON、二进制都行
SSE（Server-Sent Events）
是一种 特定标准 的“流式响应”，是“流式响应”的一种应用协议封装，规定了：
内容类型：Content-Type: text/event-stream
数据格式：必须是符合 EventStream 格式的文本流（如 data: 开头）
客户端行为：浏览器或客户端会根据标准协议自动解析事件
即流式响应是一种传输方式，而SSE是在这种传输方式的基础上，定义了协议格式和行为规范。打个比方就类似“流式响应”可以看做“公路”，“SSE”就是“按照规定走特定车道的巴士”。
实现“服务端定期推送通知消息”Demo
后端代码
# sse_server.py
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import timeapp = FastAPI()def sse_event_generator():while True:# SSE 格式，每条数据必须以 "\n\n" 结尾yield f"data: 通知来了 - {time.strftime('%X')}\n\n"time.sleep(3)@app.get("/sse")
def sse():return StreamingResponse(sse_event_generator(), media_type="text/event-stream")
前端代码
<template><div><h2>SSE 通知</h2><ul><li v-for="(msg, index) in messages" :key="index">{{ msg }}</li></ul></div><contextHolder />
</template><script setup lang="ts">
import { notification } from 'ant-design-vue';
import { onMounted, ref } from 'vue';const messages = ref([])const openNotification = (description: string) => {notification.open({message: 'SSE 通知',description: description,placement: 'topRight',duration: 2,type: 'success',});
};onMounted(() => {// 用原生 EventSource 来监听 SSE。const eventSource = new EventSource('http://localhost:5188/api/v2/sse/test')eventSource.onmessage = (event) => {messages.value.push(event.data)openNotification(event.data)}eventSource.onerror = (error) => {console.error('SSE 链接错误:', error)eventSource.close()}
})</script><style scoped>
</style>
原生 SSE 的细节规范
- Content-Type: text/event-stream
- 每条消息格式为 data: xxx\n\n
- 不能中断连接（生成器必须持续输出）
- 不得由代理缓存或破坏 chunked 传输
- 前端必须用 EventSource 来监听
进阶思考
虽然 SSE 使用起来非常高效简洁，但是在实际的业务场景中还是会遇到各种问题。例如我这里就遇到无法自动请求头、默认是 GET 请求导致的一些类似鉴权和失活的问题。下面也是一个个地来分析我遇到的问题和实际的解决方案。
EventSource 不可配置请求头
标准 EventSource 不支持设置自定义请求头。比如我想加 Authorization，因为前后端通过它实现的 JWT 鉴权来确定用户是否有权利调用接口，但是做不到。
可以将 SSE 的请求放入后端白名单中。虽然能绕过鉴权，但是日志又会缺省（不知道是哪个用户调用的）。如何选择将用户信息种到 Cookie 中又增加了复杂度（本来就是用 JWT 代替 Cookie 存储用户登录信息的）。
兼容的方案可以是在URL中加入token来鉴权和日志记录。但是又有 token 泄漏风险，即将代码复杂度转化成了安全风险。
还有一种方法就是使用 Fetch 代替 EventSource，放弃原生 EventSource，使用 fetch + ReadableStream 手动解析 SSE 协议。代码如下：
const response = await fetch("/sse", {headers: {"Authorization": "Bearer xxx"}
})
const reader = response.body.getReader()
const decoder = new TextDecoder()while (true) {const { value, done } = await reader.read()if (done) breakconst text = decoder.decode(value)console.log("收到：", text) // 你需要自己解析 `data: ...\n\n` 格式
}
看起来很好，还能解决原生 SSE 只能发送 GET 请求，不支持 POST 等请求的问题。但是这样还是会存在一些潜在的问题，即前端通知的持久化。
前端通知持久化
如果采用了 Fetch 的手动解析，会存在一个问题：当用户在任务执行之后在项目中切换到了其他页面，会导致 Fetch 在页面跳转时断开。而 SSE 的一个设计的业务场景就是用户提交长时间的耗时任务后，可以随时浏览其他内容，等任务执行完成后收到通知再回来查看结果或者执行其他任务。
前端提交耗时请求 → 后端异步处理 → 用户可以随意浏览页面 → 后端处理完毕后通过 SSE 通知前端。
这恰好卡在 EventSource 能保活，但不能带 token（受限），fetch + stream 能带 token，但连接在页面跳转时就被断开了。
业界最常用、最稳定的做法就是 “短请求 + 独立 SSE 通道”组合 —— 把“任务提交”和“任务完成通知”彻底解耦
- 用户发起耗时请求（POST）后端将任务放入后台异步任务并立即返回响应
- 在后台任务中，当任务处理完成，就通过任务 ID + 用户 ID，触发一条 SSE 消息发送给前端。
- 前端常驻 SSE 通道（全局唯一 EventSource）。这是整个应用的“通知通道”，在用户登录后建立，一直保持连接，用于接收所有异步任务完成通知。
进阶优化
可以封装一个 useNotificationSSE() composable，用于初始化全局唯一 EventSource、自动 reconnect（监听 es.onerror）、管理消息订阅（可加类型过滤、用户 ID 等）