有奖:语音产品征文挑战赛火热进行中> HOT
SSE(Server-sent Events)是?WebSocket?的一种轻量代替方案,是一种服务器端到客户端(浏览器)的单向流式消息推送协议,在 AI 生成对话等场景下较为常见。Web 函数目前已经支持通过 SSE 协议在客户端和函数运行的服务端间建立连接。

协议启用方式

SSE 协议默认支持,无需在控制台进行任何配置。

SSE 连接生命周期

在 Web 函数的 SSE 支持下, SSE 一次连接的生命周期,等同于一次函数调用请求。函数实例与连接一一对应,即同一实例在某一时刻仅处理一个 SSE 连接。在有更多连接请求发起时,将启动对应数量的实例进行处理。

操作步骤

创建函数

1. 登录 Serverless 控制台,单击左侧导航栏的函数服务
2. 在主界面上方选择期望创建函数的地域和命名空间,并单击新建,进入函数创建流程。
3. 新建函数页面,选择使用从头开始来新建函数,函数类型选择 Web 函数
4. 本文以运行环境选择 Python 3.7 为例,在函数代码中选择在线编辑,并将以下 app.py 示例代码复制粘贴至函数代码中:
import json
import time
?
from flask import Flask, Response, stream_with_context
?
app = Flask(__name__)
?
@app.route('/stream')
def stream_data():
msg = ['SSE','empowering','GPT','applications','!','Happy','chatting','!']
# 可以使用 yield 逐字返回内容
def generate_response_data():
for i,word in enumerate(msg):
json_data = json.dumps(
{'id': i, 'content': word})
yield f"data:{json_data}\\n\\n"
time.sleep(1)
return Response(stream_with_context(generate_response_data()), mimetype="text/event-stream")
?
if __name__ == '__main__':
app.run(host='0.0.0.0', port=9000)
5. 单击完成。函数创建完成后,可以在函数列表中查看。

测试函数

可以在本地终端使用 curl 工具发起 SSE 连接,示例命令如下:
curl -v -H 'Accept:text/event-stream' {函数暴露的API网关地址}/stream
返回内容如下:
> GET /release/stream HTTP/1.1
> Host: XXXXXXXXXXXXXXX.XX.apigw.tencentcs.com
> User-Agent: curl/8.0.1
> Accept: */*
> 'Accept:text/event-stream'
>
< HTTP/1.1 200 OK
< Content-Type: text/event-stream; charset=utf-8
< Transfer-Encoding: chunked
< Connection: keep-alive
< X-Api-RequestId: 22ad36c38536ee65bd07c44cb5311e1d
< Vary: Accept-Encoding
<
data:{"id": 0, "content": "SSE"}
?
data:{"id": 1, "content": "empowering"}
?
data:{"id": 2, "content": "GPT"}
?
data:{"id": 3, "content": "applications"}
?
data:{"id": 4, "content": "!"}
?
data:{"id": 5, "content": "Happy"}
?
data:{"id": 6, "content": "chatting"}
?
data:{"id": 7, "content": "!"}
?
* Connection #0 to host XXXXXXXXXXXXXXXXXXX.XX.apigw.tencentcs.com left intact
?


http://www.vxiaotou.com