有网站做淘宝客,4399曾经很火的网页游戏,wordpress 主机搬家,网站套站什么意思Python实现半双工的实时通信SSE#xff08;Server-Sent Events#xff09;
1 简介
实现实时通信一般有WebSocket、Socket.IO和SSE#xff08;Server-Sent Events#xff09;三种方法。WebSocket和Socket.IO是全双工的实时双向通信技术#xff0c;适合用于聊天和会话等Server-Sent Events
1 简介
实现实时通信一般有WebSocket、Socket.IO和SSEServer-Sent Events三种方法。WebSocket和Socket.IO是全双工的实时双向通信技术适合用于聊天和会话等但相对于SSE比较笨重SSE适合用于服务器主动向客户端实时推送数据例如用于大模型实时对话。
WebSocket是一种HTML5提供的全双工通信协议它基于TCP在客户端和服务器之间建立持久性的连接实现两者之间实时双向数据通信。
Socket.IO是一个封装了 Websocket 的实时双向数据通信库它封装了自动重连、自动检测网络状况和自动跨浏览器兼容性等。
SSEServer-Sent Events是一种利用 HTTP 协议长连接特性在服务器与客户端之间建立持久化连接实现服务器主动向客户端推送数据的半双工实时数据通信技术也被称为“事件流”Event Stream。
本文使用Python和Vue3实现SSE的实时通信现在浏览器支持EventSource不需要额外安装依赖包。
2 前端Vue3代码
script setup langts
import { onBeforeUnmount} from vuedefineProps{ msg: string }()// 定义EventSource
let eventSource: any null// 建立连接
function createSseConnect(dataId: string) {if (window.EventSource) {// 创建连接eventSource new EventSource(http://127.0.0.1:5000/sse?data_id dataId);// 接收消息eventSource.onmessage (event: MessageEvent) {console.log(onmessage: dataId : event.data)};// // 也可以使用addEventListener实现自定义事件和默认message事件// eventSource.addEventListener(message, (event: MessageEvent) {// console.log(message dataId : event.data);// }, false);// 打开连接eventSource.onopen (event: Event) {console.log(onopen: dataId : event)};// 连接出错时eventSource.onerror (event: Event) {console.log(onerror : dataId : event)};} else {console.log(浏览器不支持SSE)}
}// 组件销毁
onBeforeUnmount(() {// 关闭EventSourceif(eventSource ! null){eventSource.close()}
})/scripttemplateh1{{ msg }}/h1input typebutton value发送消息 v-on:clickcreateSseConnect(1234) //templatestyle scoped
.read-the-docs {color: #888;
}
/style
3 后端Python代码
# 导入所需的模块
import json
import time
import datetime
from flask_cors import CORS
from flask import Flask, request, Responseapp Flask(__name__)
# 解决跨域问题
CORS(app, supports_credentialsTrue)def get_data():# 获取当前时间并转换为 JSON 格式dt_ms datetime.datetime.now().strftime(%Y-%m-%d %H:%M:%S.%f)return json.dumps({time: dt_ms}, ensure_asciiFalse)app.route(/sse)
def stream():data_id request.args.get(data_id)print(data_id)return Response(eventStream(), mimetypetext/event-stream)def build_message(message: str, eventmessage):构建消息:param message: 数据消息:param event: 事件默认事件是“message”可以根据自己的需求定制事件对应前端的eventSource.addEventListener(message,(){}, false)中的message。:return:head event: event \n data:tail \n\nreturn head message taildef eventStream():id 0while True:id 1# 睡眠time.sleep(1)str_out build_message(get_data())print(str_out)# 构建迭代器yield str_outif __name__ __main__:app.run(host0.0.0.0, port5000, debugTrue)4 执行结果
