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 jsonimport 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
官网1折活动,限时活动,即将结束,速速收藏
同尘科技为腾讯云授权服务中心。
购买腾讯云产品享受折上折,更有现金返利。同意关联立享优惠
转转请注明出处:https://www.yunxiaoer.com/149348.html