跳到主要内容

SSE 协议支持

本文介绍如何在 Web 云函数中实现 SSE（Server-Sent Events）服务端推送，支持服务端向客户端单向推送实时数据流。

什么是 SSE​

「SSE（Server-Sent Events）」是一种服务器向客户端推送实时数据的技术，基于 HTTP 协议实现。与 WebSocket 不同，SSE 是单向通信，只支持服务端向客户端推送数据。

主要特点：

  • 单向推送：服务端主动向客户端推送数据，客户端只能接收
  • 基于 HTTP：使用标准 HTTP 协议，兼容性好，易于实现
  • 自动重连：客户端断线后会自动重新连接
  • 文本格式：传输的数据为文本格式（通常是 JSON）
  • 轻量实现：相比 WebSocket，实现更简单，资源占用更少

典型应用场景：

  • AI 对话流式输出
  • 实时日志推送
  • 进度更新通知
  • 服务端状态监控
  • 实时数据看板

工作原理​

协议启用​

SSE 协议在 Web 云函数中默认支持，无需在控制台进行任何额外配置即可使用。

建立连接​

  1. 客户端通过标准 HTTP 请求建立 SSE 连接
  2. 服务端返回 Content-Type: text/event-stream 响应头
  3. 连接保持打开状态，服务端持续推送数据

连接生命周期​

  • 调用对应关系：一次 SSE 连接的生命周期等同于一次函数调用请求
    • 连接建立 = 请求发起
    • 连接断开 = 请求结束
  • 实例映射：函数实例与 SSE 连接是一一对应的，同一实例在某一时刻仅处理一个 SSE 连接，新连接会启动新的实例
  • 连接保持：连接建立后，实例持续运行，通过流式响应推送数据
  • 连接结束：当 SSE 连接断开或服务端调用 end() 时，对应的函数实例停止运行

使用限制​

在使用 SSE 时，需要注意以下限制：

限制项说明
执行超时时间连接持续时间受函数最大运行时长限制
并发连接每个连接启动独立实例，受账户并发配额限制
浏览器连接数限制同一域名下浏览器对 SSE 连接数有限制（通常为 6 个）
数据格式只能传输文本数据，复杂数据需序列化为 JSON

操作步骤​

步骤1：编写服务端代码​

SSE 协议默认支持，无需在控制台开启。根据您使用的编程语言和框架，编写 SSE 服务端代码。

安装依赖

安装 Express 框架：

npm install express

在 package.json 中添加依赖：

{
	"dependencies": {
		"express": "^4.18.0"
	}
}

编写服务端代码

使用 Express 实现 SSE 流式响应：

const express = require('express');
const app = express();

// SSE 路由
app.get('/stream', (req, res) => {
	// 设置 SSE 响应头
	res.setHeader('Content-Type', 'text/event-stream');
	res.setHeader('Cache-Control', 'no-cache');
	res.setHeader('Connection', 'keep-alive');

	console.log('新的 SSE 连接建立');

	// 流式推送数据
	const msg = ['SSE', 'empowering', 'GPT', 'applications', '!', 'Happy', 'chatting', '!'];

	let index = 0;
	const intervalId = setInterval(() => {
		if (index < msg.length) {
			// 发送 SSE 消息
			const data = {
				id: index,
				content: msg[index],
			};
			res.write(`data: ${JSON.stringify(data)}\n\n`);
			index++;
		} else {
			// 完成推送，关闭连接
			clearInterval(intervalId);
			res.end();
		}
	}, 1000);

	// 客户端断开连接时清理
	req.on('close', () => {
		clearInterval(intervalId);
		console.log('SSE 连接已关闭');
	});
});

// 启动服务，监听 9000 端口
app.listen(9000, () => {
	console.log('SSE 服务已启动，监听端口 9000');
});

步骤2：部署云函数​

完成代码编写后，部署云函数：

  • 在控制台点击「部署」按钮
  • 或使用 CloudBase CLI：tcb fn deploy --httpFn

步骤3：客户端连接测试​

使用浏览器或命令行工具连接 SSE 服务：

使用 curl 测试

curl -v -H 'Accept:text/event-stream' https://your-function.run.tcloudbase.com/stream

预期返回结果

服务器将以流的形式分块返回数据，每条数据间隔约 1 秒：

data: {"id": 0, "content": "SSE"}

data: {"id": 1, "content": "empowering"}

data: {"id": 2, "content": "GPT"}

...

使用浏览器 JavaScript 测试

// 创建 SSE 连接
const eventSource = new EventSource('https://your-function.run.tcloudbase.com/stream');

eventSource.onmessage = (event) => {
	console.log('收到消息:', event.data);

	try {
		const data = JSON.parse(event.data);
		console.log('解析数据:', data);
	} catch (e) {
		// 处理非 JSON 数据
	}
};

eventSource.onerror = (error) => {
	console.error('SSE 错误:', error);
};

高级用法​

1. AI 对话流式输出​

实现类似 ChatGPT 的流式对话效果：

const express = require('express');
const app = express();

app.use(express.json());

// AI 对话流式输出
app.post('/chat', async (req, res) => {
	// 设置 SSE 响应头
	res.setHeader('Content-Type', 'text/event-stream');
	res.setHeader('Cache-Control', 'no-cache');
	res.setHeader('Connection', 'keep-alive');

	const { message } = req.body;

	// 发送开始事件
	res.write(`event: start\ndata: ${JSON.stringify({ status: '开始生成回复' })}\n\n`);

	// 模拟 AI 生成回复（实际应调用 AI 模型 API）
	const reply = '这是 AI 生成的回复内容，会逐字推送给用户。';

	for (let i = 0; i < reply.length; i++) {
		const data = {
			content: reply[i],
			index: i,
		};
		res.write(`event: message\ndata: ${JSON.stringify(data)}\n\n`);

		// 模拟生成延迟
		await new Promise((resolve) => setTimeout(resolve, 50));
	}

	// 发送完成事件
	res.write(
		`event: done\ndata: ${JSON.stringify({ status: '生成完成', totalLength: reply.length })}\n\n`
	);
	res.end();
});

app.listen(9000);

2. 实时进度推送​

推送任务执行进度：

const express = require('express');
const app = express();

app.get('/progress', async (req, res) => {
	res.setHeader('Content-Type', 'text/event-stream');
	res.setHeader('Cache-Control', 'no-cache');
	res.setHeader('Connection', 'keep-alive');

	const totalSteps = 10;

	for (let step = 1; step <= totalSteps; step++) {
		const data = {
			step,
			total: totalSteps,
			percent: Math.round((step / totalSteps) * 100),
			message: `正在处理第 ${step} 步...`,
		};

		res.write(`event: progress\ndata: ${JSON.stringify(data)}\n\n`);

		// 模拟任务执行
		await new Promise((resolve) => setTimeout(resolve, 1000));
	}

	// 任务完成
	res.write(
		`event: complete\ndata: ${JSON.stringify({ status: '任务完成', timestamp: Date.now() })}\n\n`
	);
	res.end();
});

app.listen(9000);

3. 实时日志推送​

推送服务端日志到客户端：

const express = require('express');
const app = express();

// 模拟日志队列
const logQueue = [];

// 模拟生成日志
setInterval(() => {
	logQueue.push({
		level: ['info', 'debug', 'warn'][Math.floor(Math.random() * 3)],
		message: `日志消息 ${Date.now()}`,
		timestamp: Date.now(),
	});

	// 限制队列大小
	if (logQueue.length > 100) {
		logQueue.shift();
	}
}, 2000);

app.get('/logs', (req, res) => {
	res.setHeader('Content-Type', 'text/event-stream');
	res.setHeader('Cache-Control', 'no-cache');
	res.setHeader('Connection', 'keep-alive');

	console.log('开始推送日志');

	// 定期推送日志
	const intervalId = setInterval(() => {
		if (logQueue.length > 0) {
			const logs = logQueue.splice(0, 5); // 每次推送最多 5 条

			logs.forEach((log) => {
				res.write(`event: log\ndata: ${JSON.stringify(log)}\n\n`);
			});
		}
	}, 1000);

	// 客户端断开时清理
	req.on('close', () => {
		clearInterval(intervalId);
		console.log('停止推送日志');
	});
});

app.listen(9000);

4. 服务端状态监控​

实时推送服务器状态指标：

const express = require('express');
const app = express();

app.get('/metrics', (req, res) => {
	res.setHeader('Content-Type', 'text/event-stream');
	res.setHeader('Cache-Control', 'no-cache');
	res.setHeader('Connection', 'keep-alive');
	res.setHeader('X-Monitoring-Session', `session-${Date.now()}`);

	console.log('开始推送监控数据');

	// 定期推送监控数据
	const intervalId = setInterval(() => {
		const metrics = {
			cpu: Math.random() * 100,
			memory: Math.random() * 100,
			activeConnections: Math.floor(Math.random() * 1000),
			timestamp: Date.now(),
		};

		res.write(`event: metrics\ndata: ${JSON.stringify(metrics)}\n\n`);
	}, 2000);

	// 5 分钟后超时
	const timeoutId = setTimeout(() => {
		clearInterval(intervalId);
		res.write(`event: timeout\ndata: ${JSON.stringify({ message: '监控会话超时' })}\n\n`);
		res.end();
	}, 300000);

	// 客户端断开时清理
	req.on('close', () => {
		clearInterval(intervalId);
		clearTimeout(timeoutId);
		console.log('停止推送监控数据');
	});
});

app.listen(9000);

客户端连接示例​

// 创建 SSE 连接
const eventSource = new EventSource('https://your-service.run.tcloudbase.com/stream');

// 监听默认消息事件
eventSource.onmessage = (event) => {
	console.log('收到消息:', event.data);

	try {
		const data = JSON.parse(event.data);
		console.log('解析的数据:', data);
	} catch (e) {
		// 处理非 JSON 数据
	}
};

// 监听自定义事件
eventSource.addEventListener('progress', (event) => {
	const data = JSON.parse(event.data);
	console.log(`进度: ${data.percent}%`);
});

eventSource.addEventListener('done', (event) => {
	console.log('任务完成');
	eventSource.close(); // 关闭连接
});

// 连接打开
eventSource.onopen = () => {
	console.log('SSE 连接已建立');
};

// 错误处理
eventSource.onerror = (error) => {
	console.error('SSE 错误:', error);
	// 可选择关闭连接
	// eventSource.close();
};

// 手动关闭连接
function closeConnection() {
	eventSource.close();
	console.log('SSE 连接已关闭');
}

SSE 消息格式​

SSE 协议使用特定的文本格式发送消息，每条消息由一个或多个字段组成：

event: message
data: {"content": "Hello"}
id: 123
retry: 1000

标准字段​

字段说明是否必需
data消息数据内容是
event事件类型（默认为 message）否
id事件 ID，用于客户端重连时恢复否
retry重连间隔时间（毫秒）否
: 注释注释行，客户端会忽略（用于保持连接）否

消息示例​

基本消息

data: Hello World

JSON 数据

data: {"message": "Hello", "timestamp": 1234567890}

带事件类型

event: notification
data: {"type": "info", "content": "新消息"}

多行数据

data: 第一行
data: 第二行
data: 第三行

完整消息

event: update
id: 123
retry: 1000
data: {"status": "success"}

常见问题​

1. 连接无法建立怎么办？​

  • 确认云函数已正确部署和运行
  • 检查响应头是否设置了 Content-Type: text/event-stream
  • 确认使用正确的函数访问地址
  • 查看云函数日志是否有错误信息
  • 检查客户端浏览器是否支持 SSE（IE 不支持）

2. 连接频繁断开怎么办？​

  • 增加函数执行超时时间配置
  • 实现心跳机制，定期发送注释行保持连接：res.write(': keepalive\n\n')
  • 检查客户端网络是否稳定
  • 查看云函数日志，确认是否有异常错误导致连接断开
  • 客户端实现自动重连机制（EventSource 默认支持）

3. 如何实现断线重连？​

EventSource 客户端会自动重连，服务端可以通过 id 和 retry 字段配合：

服务端发送 ID 和重连间隔

res.write(`id: ${messageId}\nretry: 3000\ndata: ${JSON.stringify(data)}\n\n`);

客户端获取最后的事件 ID

const eventSource = new EventSource('https://your-service.run.tcloudbase.com/stream');

eventSource.onopen = () => {
	console.log('连接已建立');
};

eventSource.onerror = () => {
	console.log('连接断开，自动重连中...');
};

4. 如何处理大量并发连接？​

  • 实例扩展：云函数会自动为每个 SSE 连接创建独立实例
  • 连接数限制：注意浏览器同域名连接数限制（通常为 6 个）
  • 状态管理：使用 Redis 等外部存储共享连接状态
  • 消息队列：使用消息队列处理广播和异步消息
  • 资源监控：关注函数实例数量和资源使用情况

5. SSE 和 WebSocket 如何选择？​

选择 SSE 的场景：

  • 只需要服务端向客户端推送数据
  • 追求简单实现和快速开发
  • 需要自动重连功能
  • 兼容性要求较高（HTTP 协议）

选择 WebSocket 的场景：

  • 需要双向实时通信
  • 需要传输二进制数据
  • 需要更低的延迟和更高的效率
  • 客户端需要频繁向服务端发送数据

6. 如何发送心跳保持连接？​

定期发送注释行（以 : 开头）保持连接活跃：

const express = require('express');
const app = express();

app.get('/stream', (req, res) => {
	res.setHeader('Content-Type', 'text/event-stream');
	res.setHeader('Cache-Control', 'no-cache');
	res.setHeader('Connection', 'keep-alive');

	// 每 30 秒发送心跳
	const heartbeatInterval = setInterval(() => {
		res.write(': heartbeat\n\n');
	}, 30000);

	// 发送实际数据
	const dataInterval = setInterval(() => {
		res.write(`data: ${JSON.stringify({ timestamp: Date.now() })}\n\n`);
	}, 5000);

	req.on('close', () => {
		clearInterval(heartbeatInterval);
		clearInterval(dataInterval);
	});
});

app.listen(9000);

SSE vs WebSocket​

特性SSEWebSocket
通信方式单向（服务端→客户端）双向（服务端↔客户端）
协议HTTPWebSocket 协议
数据格式文本（通常 JSON）文本或二进制
自动重连是否（需手动实现）
浏览器兼容性好（IE 除外）好
实现复杂度简单相对复杂
资源占用较低较高
配置要求无需配置需要在控制台开启
适用场景单向数据推送实时双向通信

选择建议：

  • 只需要服务端推送数据：选择 SSE
  • 需要双向实时通信：选择 WebSocket
  • 追求简单实现：选择 SSE
  • 需要传输二进制数据：选择 WebSocket

相关文档​