FastAPI实现后端实时推送:告别轮询,拥抱SSE与WebSocket

FastAPI实现后端实时推送:告别轮询,拥抱SSE与WebSocket

本文旨在解决%ignore_a_1%频繁轮询后端以获取实时状态更新的低效问题。我们将探讨两种高效的后端数据推送机制:server-sent events (sse) 和 websockets。重点介绍它们的工作原理、fastapi中的实现方式以及前端如何接收数据,并根据实际应用场景,提供选择这两种技术的指导,以实现从后端向前端的事件驱动型实时通信。

在现代Web应用中,实时数据更新是提升用户体验的关键。传统的做法是前端定时向后端发送请求(即轮询)以检查数据是否有更新。然而,当数据更新不频繁时,这种方式会造成大量的无效请求,浪费服务器资源和网络带宽,尤其是在硬件状态等可能长时间保持不变的场景下,其低效性尤为突出。为了解决这一问题,我们需要一种机制,让后端能够在数据发生变化时主动将信息推送给前端,而不是等待前端的请求。

实时数据推送机制概述

实现后端向前端实时推送数据主要有两种主流技术:Server-Sent Events (SSE) 和 WebSockets。它们都允许服务器在数据可用时立即发送给客户端,从而避免了低效的轮询。

1. Server-Sent Events (SSE)

Server-Sent Events 是一种基于HTTP的单向通信技术,允许服务器向客户端推送事件流。它通过一个持久的HTTP连接工作,服务器可以随时向客户端发送文本事件。

工作原理:当客户端发起一个SSE连接时,服务器会保持这个连接打开,并以 text/event-stream 的MIME类型发送数据。数据以特定格式(data: your_message\n\n)发送,客户端的浏览器会自动解析这些事件。

优点:

ViiTor实时翻译 ViiTor实时翻译

AI实时多语言翻译专家!强大的语音识别、AR翻译功能。

ViiTor实时翻译 116 查看详情 ViiTor实时翻译 简单易用: 基于HTTP,无需特殊协议或复杂库,浏览器原生支持 EventSource API。自动重连: 客户端的 EventSource 会在连接断开时自动尝试重新连接。单向通信: 适用于服务器只向客户端推送数据的场景,如实时通知、股票行情、硬件状态更新等。防火墙友好: 使用标准HTTP协议,通常不会被防火墙阻拦。

FastAPI 实现 SSE:在FastAPI中,可以使用 StreamingResponse 结合一个异步生成器来创建SSE端点。当硬件状态发生变化时,后端可以将最新的状态作为事件发送给所有订阅的客户端。

from fastapi import FastAPI, Requestfrom fastapi.responses import StreamingResponseimport asyncioimport jsonapp = FastAPI()# 模拟硬件状态hardware_status = {"temperature": 25.0, "humidity": 60, "power_on": True}# 存储待发送的事件event_queue = asyncio.Queue()# 模拟硬件状态更新(在实际应用中,这会由硬件监控脚本触发)async def simulate_hardware_updates():    while True:        await asyncio.sleep(5) # 每5秒模拟一次状态更新        new_temperature = hardware_status["temperature"] + 0.5        new_humidity = hardware_status["humidity"] + (1 if new_temperature > 27 else -1)        # 假设只有温度或湿度变化才推送        if new_temperature != hardware_status["temperature"] or new_humidity != hardware_status["humidity"]:            hardware_status["temperature"] = round(new_temperature, 2)            hardware_status["humidity"] = round(new_humidity, 2)            print(f"Hardware status updated: {hardware_status}")            # 将更新后的状态放入事件队列            event_data = {"status": hardware_status, "timestamp": asyncio.time()}            await event_queue.put(json.dumps(event_data))@app.on_event("startup")async def startup_event():    asyncio.create_task(simulate_hardware_updates())@app.get("/events")async def sse_endpoint(request: Request):    async def event_generator():        while True:            # 检查客户端是否断开连接            if await request.is_disconnected():                print("Client disconnected from SSE.")                break            # 从队列获取事件            event_data = await event_queue.get()            yield f"data: {event_data}\n\n"            # 确保在没有事件时不会阻塞太久,可以加入一个短时间的延迟            await asyncio.sleep(0.1)     return StreamingResponse(event_generator(), media_type="text/event-stream")# 额外的端点,用于手动触发状态更新(可选,用于测试)@app.post("/update_status")async def update_status(new_temp: float = 26.0, new_hum: int = 65):    hardware_status["temperature"] = new_temp    hardware_status["humidity"] = new_hum    event_data = {"status": hardware_status, "timestamp": asyncio.time()}    await event_queue.put(json.dumps(event_data))    return {"message": "Status updated and event queued."}

前端 (React) 接收 SSE:前端使用 EventSource API来监听来自 /events 端点的事件。

import React, { useEffect, useState } from 'react';function HardwareStatus() {  const [status, setStatus] = useState(null);  const [error, setError] = useState(null);  useEffect(() => {    // 创建 EventSource 实例    const eventSource = new EventSource('http://localhost:8000/events'); // 替换为你的FastAPI地址    // 监听 'message' 事件,这是默认的事件类型    eventSource.onmessage = (event) => {      try {        const data = JSON.parse(event.data);        setStatus(data.status);        setError(null); // 清除之前的错误        console.log("Received SSE event:", data);      } catch (e) {        console.error("Failed to parse SSE data:", e);        setError("Failed to parse data.");      }    };    // 监听 'open' 事件,表示连接已建立    eventSource.onopen = () => {      console.log('SSE connection opened.');    };    // 监听 'error' 事件    eventSource.onerror = (error) => {      console.error('SSE error:', error);      eventSource.close(); // 发生错误时关闭连接      setError("SSE connection error. Retrying...");      // EventSource 会自动尝试重连,但你也可以在这里自定义重连逻辑    };    // 组件卸载时关闭连接    return () => {      eventSource.close();      console.log('SSE connection closed.');    };  }, []); // 仅在组件挂载时运行一次  if (error) {    return 
Error: {error}
; } if (!status) { return
Connecting to hardware status updates...
; } return (

Hardware Status

Temperature: {status.temperature}°C

Humidity: {status.humidity}%

Power On: {status.power_on ? 'Yes' : 'No'}

);}export default HardwareStatus;

2. WebSockets

WebSockets 提供了一个全双工的通信通道,允许客户端和服务器之间进行双向、低延迟的实时数据交换。与SSE不同,WebSockets在握手后建立一个持久的TCP连接,而不是基于HTTP请求-响应模型。

工作原理:客户端通过一个特殊的HTTP握手请求升级到WebSocket协议。一旦握手成功,连接就升级为WebSocket,客户端和服务器可以独立地发送和接收数据帧。

优点:

双向通信: 客户端和服务器可以同时发送和接收消息,适用于聊天应用、多人协作等场景。低延迟: 建立连接后,数据传输开销小,延迟低。更灵活: 可以传输二进制数据,支持更复杂的消息格式。

FastAPI 实现 WebSockets:FastAPI通过 websocket 依赖提供了对WebSockets的良好支持。

from fastapi import FastAPI, WebSocket, WebSocketDisconnectimport asyncioimport jsonapp = FastAPI()# 模拟硬件状态hardware_status_ws = {"temperature": 25.0, "humidity": 60, "power_on": True}# 存储所有活跃的WebSocket连接active_connections: list[WebSocket] = []# 模拟硬件状态更新(在实际应用中,这会由硬件监控脚本触发)async def simulate_hardware_updates_ws():    while True:        await asyncio.sleep(5) # 每5秒模拟一次状态更新        new_temperature = hardware_status_ws["temperature"] + 0.5        new_humidity = hardware_status_ws["humidity"] + (1 if new_temperature > 27 else -1)        if new_temperature != hardware_status_ws["temperature"] or new_humidity != hardware_status_ws["humidity"]:            hardware_status_ws["temperature"] = round(new_temperature, 2)            hardware_status_ws["humidity"] = round(new_humidity, 2)            print(f"Hardware status updated (WS): {hardware_status_ws}")            # 向所有连接的客户端广播更新            message = json.dumps({"status": hardware_status_ws, "timestamp": asyncio.time()})            for connection in active_connections:                try:                    await connection.send_text(message)                except RuntimeError as e:                    print(f"Error sending to WebSocket client: {e}")                    # 可以在这里处理断开的连接,例如从 active_connections 中移除@app.on_event("startup")async def startup_event_ws():    asyncio.create_task(simulate_hardware_updates_ws())@app.websocket("/ws")async def websocket_endpoint(websocket: WebSocket):    await websocket.accept()    active_connections.append(websocket)    print(f"New WebSocket connection: {websocket.client}")    # 首次连接时发送当前状态    initial_status = json.dumps({"status": hardware_status_ws, "timestamp": asyncio.time()})    await websocket.send_text(initial_status)    try:        while True:            # 可以在这里接收客户端发送的消息(如果需要双向通信)            # data = await websocket.receive_text()            # print(f"Received from client: {data}")            # await websocket.send_text(f"Message text was: {data}")            await asyncio.sleep(0.1) # 保持连接活跃,避免CPU空转    except WebSocketDisconnect:        active_connections.remove(websocket)        print(f"WebSocket client disconnected: {websocket.client}")    except Exception as e:        print(f"WebSocket error: {e}")        active_connections.remove(websocket)

前端 (React) 接收 WebSockets:前端使用 WebSocket API来建立和管理连接。

import React, { useEffect, useState, useRef } from 'react';function HardwareStatusWS() {  const [status, setStatus] = useState(null);  const [error, setError] = useState(null);  const ws = useRef(null); // 使用ref来存储WebSocket实例  useEffect(() => {    // 建立 WebSocket 连接    ws.current = new WebSocket('ws://localhost:8000/ws'); // 替换为你的FastAPI地址    ws.current.onopen = () => {      console.log('WebSocket connection opened.');      setError(null); // 清除之前的错误    };    ws.current.onmessage = (event) => {      try {        const data = JSON.parse(event.data);        setStatus(data.status);        console.log("Received WebSocket message:", data);      } catch (e) {        console.error("Failed to parse WebSocket data:", e);        setError("Failed to parse data.");      }    };    ws.current.onclose = (event) => {      console.log('WebSocket connection closed:', event.code, event.reason);      setError("WebSocket connection closed. Reconnecting...");      // 可以实现重连逻辑      setTimeout(() => {        // Simple reconnect logic, consider more robust solutions for production        if (ws.current && ws.current.readyState === WebSocket.CLOSED) {            console.log("Attempting to reconnect WebSocket...");            ws.current = null; // Clear old instance            // Trigger effect to re-establish connection            // This is a simple way, often a dedicated reconnect function is better            // For simplicity, we'll let the effect re-run if dependencies change, or manually call a reconnect function            // For now, simply setting ws.current to null and letting the next render potentially re-trigger setup is too indirect.            // A more direct approach:            // ws.current = new WebSocket('ws://localhost:8000/ws'); // Re-initiate connection            // And then re-attach handlers, or better, wrap this in a function.        }      }, 3000); // 3秒后尝试重连    };    ws.current.onerror = (error) => {      console.error('WebSocket error:', error);      setError("WebSocket connection error.");    };    // 组件卸载时关闭连接    return () => {      if (ws.current) {        ws.current.close();        console.log('WebSocket connection cleaned up.');      }    };  }, []); // 仅在组件挂载时运行一次  // 示例:向服务器发送消息(如果需要双向通信)  // const sendMessage = () => {  //   if (ws.current && ws.current.readyState === WebSocket.OPEN) {  //     ws.current.send(JSON.stringify({ message: "Hello from client!" }));  //   }  // };  if (error) {    return 
Error: {error}
; } if (!status) { return
Connecting to hardware status updates via WebSocket...
; } return (

Hardware Status (WebSocket)

Temperature: {status.temperature}°C

Humidity: {status.humidity}%

Power On: {status.power_on ? 'Yes' : 'No'}

{/* */}
);}export default HardwareStatusWS;

SSE 与 WebSockets 的选择

在实际应用中,选择SSE还是WebSockets取决于具体的业务需求:

SSE (Server-Sent Events):

推荐场景: 当你只需要从服务器向客户端单向推送数据时,例如实时通知、股票报价、新闻推送、日志流、以及本例中硬件状态更新(客户端不需要频繁发送消息给服务器)。优势: 实现简单,浏览器原生支持自动重连,基于HTTP协议,防火墙友好。劣势: 只能单向通信,每次事件传输的数据量相对较小。

WebSockets:

推荐场景: 当你需要客户端和服务器之间进行双向、低延迟的实时通信时,例如聊天应用、在线游戏、实时协作文档、需要客户端频繁发送控制指令的物联网应用。优势: 全双工通信,协议开销小,延迟低,可以传输二进制数据。劣势: 相较于SSE,实现略复杂,需要处理连接管理、心跳、重连逻辑等,可能需要额外的库。

针对本案例(硬件状态更新,可能长时间无变化):由于硬件状态更新属于服务器向客户端的单向推送,且可能长时间处于空闲状态,SSE是更推荐的选择。它的实现更简单,并且内置的自动重连机制能够很好地处理连接中断的情况,而无需客户端进行复杂的重连逻辑。WebSockets虽然也能实现,但对于这种单向推送的场景,其双向通信的能力并没有被充分利用,反而增加了实现的复杂性。

注意事项与最佳实践

连接管理: 对于WebSockets,需要妥善管理 active_connections 列表,确保在客户端断开连接时及时移除,避免向已断开的连接发送数据导致错误。SSE的 StreamingResponse 也会在客户端断开时抛出 RequestDisconnected 异常,需要捕获并处理。消息队列: 在高并发场景下,直接在FastAPI应用中维护 event_queue 或广播到 active_connections 可能不够高效。可以考虑引入外部消息队列(如Redis Pub/Sub、Kafka、RabbitMQ)来解耦事件生产者和消费者,实现更强大的广播和扩展能力。心跳机制: 对于长时间不发送数据的SSE或WebSocket连接,某些代理或负载均衡器可能会因为不活跃而断开连接。可以定期发送一个空事件(SSE)或心跳消息(WebSocket)来保持连接活跃。错误处理与重连: 客户端应始终包含健壮的错误处理和重连逻辑。SSE内置了自动重连,但WebSocket需要手动实现。安全性: 实时通信端点也应进行认证和授权,确保只有合法的客户端才能接收或发送数据。可以结合FastAPI的依赖注入系统实现。可伸缩性: 当客户端数量增加时,单个FastAPI实例可能无法处理所有实时连接。可以考虑使用负载均衡器将连接分发到多个FastAPI实例,并结合消息队列实现跨实例的事件广播。

总结

通过采用SSE或WebSockets,我们可以彻底告别低效的轮询机制,实现后端数据向前端的实时、事件驱动型推送。对于本例中硬件状态更新这种以服务器向客户端单向推送为主,且可能长时间空闲的场景,SSE因其简洁性和内置的自动重连特性而成为更优的选择。而当需要客户端与服务器进行频繁双向通信时,WebSockets则能提供更强大的支持。理解这两种技术的特点并根据实际需求做出明智的选择,是构建高效实时Web应用的关键。

以上就是FastAPI实现后端实时推送:告别轮询,拥抱SSE与WebSocket的详细内容,更多请关注创想鸟其它相关文章!

版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。
如发现本站有涉嫌抄袭侵权/违法违规的内容, 请发送邮件至 chuangxiangniao@163.com 举报,一经查实,本站将立刻删除。
发布者:程序猿,转转请注明出处:https://www.chuangxiangniao.com/p/580659.html

(0)
打赏 微信扫一扫 微信扫一扫 支付宝扫一扫 支付宝扫一扫
上一篇 2025年11月10日 10:46:38
下一篇 2025年11月10日 10:47:27

相关推荐

  • C# web api返回类型设置为json的两种方法

    web api写api接口时默认返回的是把你的对象序列化后以xml形式返回,那么怎样才能让其返回为json呢,下面就介绍两种方法: 方法一:(改配置法)  找到global.asax文件,在application_start()方法中添加一句:  GlobalConfiguration.Config…

    好文分享 2025年12月17日
    000
  • RSS订阅中的作者信息格式

    RSS和Atom中作者信息通过或标签标识,包含姓名、邮箱及网站链接,支持多作者;正确设置有助于提升内容可信度、便于追踪与SEO。 RSS订阅中的作者信息格式,主要用于标识文章的作者,让读者知道是谁写的,方便追踪特定作者的内容。格式通常包含作者姓名、邮箱,有时还会包含作者的网站链接。 作者信息的常见格…

    2025年12月17日
    000
  • XML中如何获取根节点属性_XML获取根节点属性的操作步骤

    XML根节点有且仅有一个,可包含属性;2. Python用ET.parse解析,root.get(“属性名”)获取属性值;3. JavaScript用DOMParser解析,xmlDoc.documentElement获取根节点,getAttribute读取属性;4. Jav…

    2025年12月17日
    000
  • XML中如何去除空节点_XML去除空节点的实用方法

    答案:可通过XSLT、Python脚本或命令行工具去除XML空节点。使用XSLT模板递归复制非空节点;Python的lxml库遍历并删除无文本、无子节点、无属性的元素;XMLStarlet命令行工具执行XPath表达式快速清理空标签,处理前需明确定义空节点并备份原文件。            &lt…

    2025年12月17日
    000
  • XML中如何生成XML报表模板_XML生成XML报表模板的方法与示例

    利用XSLT、编程语言或模板引擎可生成XML报表模板:1. XSLT将源XML转换为结构化报表;2. Python等语言通过DOM操作动态构建XML;3. Jinja2等模板引擎支持变量与逻辑控制,实现灵活输出。 在XML中生成XML报表模板,实际上是指利用XML的结构化特性设计一个可复用的数据模板…

    2025年12月17日
    000
  • XML中如何解压XML字符串_XML解压XML字符串的操作方法

    先解压再解析XML。C#用GZipStream解压字节流并转字符串,Java用GZIPInputStream或InflaterInputStream读取压缩数据,结合StreamReader或BufferedReader还原为明文XML后,交由XDocument或DocumentBuilder解析;…

    2025年12月17日
    000
  • XML中如何转换XML编码格式_XML转换XML编码格式的方法与技巧

    正确识别并统一XML文件的编码声明与实际编码是解决解析错误的关键,可通过编辑器、命令行或编程方式(如Python脚本)进行转换,确保内容、声明和保存编码一致,避免乱码。 配合XSLT处理器(如Saxon),可实现内容转换的同时完成编码标准化。 基本上就这些。关键点是确保文件内容、XML声明、保存编码…

    2025年12月17日
    000
  • XML中如何判断节点是否存在_XML判断节点存在性的技巧与方法

    使用XPath或find方法判断XML节点是否存在,若返回结果为空则节点不存在,结合attrib检查属性,并区分节点存在与文本内容是否为空。 在处理XML文档时,判断某个节点是否存在是一个常见需求。无论是解析配置文件、处理接口返回数据,还是进行数据校验,准确判断节点是否存在可以避免程序出错。以下是几…

    2025年12月17日
    000
  • XML中如何生成XML文档_XML生成XML文档的详细操作方法

    使用Python、Java和JavaScript均可生成XML文档。Python通过ElementTree创建根节点与子节点并写入文件;Java利用DOM API构建元素层级并转换输出;JavaScript借助xmlbuilder库链式生成结构化XML,均需注意命名规范及特殊字符处理。 在程序开发中…

    2025年12月17日
    000
  • XML中如何遍历所有节点_XML遍历节点的操作方法与实践

    使用Python的ElementTree和Java的DOM均可递归遍历XML所有节点,前者通过iter()方法访问每个元素,后者利用NodeList递归处理子节点,实现信息提取或修改。 在处理XML数据时,经常需要遍历所有节点以提取信息或进行修改。实现这一目标的方法取决于使用的编程语言和解析库,但核…

    2025年12月17日
    000
  • XML中如何检查节点顺序_XML检查节点顺序的方法与技巧

    使用XPath、DOM解析、XSD约束和断言工具可检查XML节点顺序。首先通过XPath的position()函数验证节点位置,如//data/item[@type=’A’ and position()=1];其次用Python等语言解析DOM并比对实际与预期顺序;再者利用X…

    2025年12月17日
    000
  • 如何优化XML网络传输

    优化XML网络传输需从压缩、结构精简和协议升级入手。首先,Gzip压缩可减少60%-80%数据量;其次,简化标签名、去除冗余命名空间与空白字符能降低XML“体重”;再者,采用SAX或XMLPullParser流式解析替代DOM,可显著提升大文件处理效率;同时,预编译XPath/XSLT、缓存解析结果…

    2025年12月17日
    000
  • RSS源如何实现内容推荐

    要实现RSS%ignore_a_1%,需在RSS数据基础上构建智能推荐系统。首先通过feedparser等工具抓取并解析RSS内容,提取标题、摘要、发布时间等信息,并存储到数据库中;对于仅提供片段的源,可结合Web Scraping技术获取全文。随后利用NLP技术对内容进行处理,包括分词、去停用词、…

    2025年12月17日
    000
  • 如何用XML表示时间序列数据

    XML通过层级结构和属性封装时间戳与数值,适合表示含丰富元数据和不规则采样的时间序列数据,便于跨系统交换;其优势在于自描述性、可扩展性和平台无关性,但存在冗余大、解析慢等问题,海量数据时不如二进制格式或专用数据库高效。 在XML中表示时间序列数据,核心在于利用其层级结构和属性来封装每个时间点的数据值…

    2025年12月17日
    000
  • XML中如何使用XSLT样式转换_XML使用XSLT样式转换XML的方法与示例

    XSLT通过样式表将XML转换为HTML等格式,需准备XML源文件、编写XSLT规则并使用处理器执行转换。 在XML中使用XSLT进行样式转换,主要是通过编写XSLT样式表来定义XML数据的输出格式。XSLT(Extensible Stylesheet Language Transformation…

    2025年12月17日
    000
  • XML中如何反序列化XML对象_XML反序列化XML对象的操作方法

    答案:C#和Java可通过XmlSerializer和JAXB实现XML反序列化,需定义匹配类并使用特性/注解映射字段,确保无参构造函数和正确命名空间,最终将XML数据转换为对象。 在处理XML数据时,反序列化是将XML格式的数据转换为程序中的对象的过程。这一操作广泛应用于配置读取、网络通信和数据存…

    2025年12月17日
    000
  • RSS阅读器如何开发?核心功能有哪些?

    答案:开发RSS阅读器需实现订阅管理、内容抓取解析、展示与同步功能,采用Node.js或Python等技术栈,支持OPML导入、定时更新、离线缓存,并防范XXE攻击,提升用户体验。 RSS阅读器的开发核心在于抓取、解析和展示网站的RSS订阅源内容。这类工具帮助用户集中浏览多个网站的更新,无需逐个访问…

    2025年12月17日
    000
  • XML文档对象模型如何构建?编程接口介绍。

    DOM将XML文档加载到内存中构建树形结构,便于遍历、查询和修改。01. 它将元素、属性、文本等视为节点,形成以document为根的树。02. 常见节点类型包括Element、Attribute、Text、Comment和Document。03. 核心API支持创建、查找、添加、删除节点及获取属性…

    2025年12月17日
    000
  • XML中如何反序列化XML为对象_XML反序列化XML为对象的操作方法

    答案:XML反序列化是将XML数据转换为程序对象的过程,C#使用XmlSerializer类,Java使用JAXB实现。需定义与XML结构匹配的类,添加相应特性或注解,确保无参构造函数存在,通过Deserialize或unmarshal方法完成转换,注意标签名匹配、命名空间和集合类型处理,避免解析失…

    2025年12月17日
    000
  • RSS中的skipHours元素作用

    skipHours是RSS中用于优化更新频率的元素,发布者可通过它指定某些小时段让订阅客户端暂停检查更新,以减少无效请求、降低服务器负载。 RSS中的skipHours元素,说白了,就是发布者在告诉订阅者(或者说,订阅客户端):在某些特定的小时段里,你暂时不用来检查我的更新了。它提供了一种精细化的机…

    2025年12月17日
    000

发表回复

登录后才能评论
关注微信