本文探讨了在fastapi后端向react前端推送实时硬件状态更新的有效方法，旨在解决传统轮询机制在状态不常变化时效率低下的问题。我们将重点介绍两种事件驱动的通信模式：server-sent events (sse) 和 websocket，并分析其适用场景，提供实现示例，帮助开发者构建响应更及时、资源消耗更低的实时应用。
在现代Web应用开发中，实时数据更新是提升用户体验的关键。尤其当涉及到硬件状态监控这类场景时，前端需要及时反映后端的变化。然而，传统的客户端轮询（即前端定时向后端发送请求查询数据）机制在数据不常变化时效率低下，会造成不必要的网络流量和服务器资源消耗。为了解决这一问题，事件驱动的通信模式应运而生，其中Server-Sent Events (SSE) 和 WebSocket 是两种主流且高效的解决方案。
当后端需要主动向前端推送数据，而非等待前端请求时，我们需要建立一种持久的连接或订阅机制。这正是SSE和WebSocket所擅长的领域。
Server-Sent Events 是一种基于HTTP协议的单向通信技术，允许服务器持续地向客户端推送数据。它利用了HTTP长连接，服务器可以在连接打开的情况下，通过特定的MIME类型（text/event-stream）发送一系列事件。
适用场景： SSE特别适合于那些数据流向主要是从服务器到客户端的场景，例如实时股价更新、新闻推送、日志监控或本文所讨论的硬件状态更新（当状态变化不频繁，且主要由服务器发起时）。它的优势在于实现相对简单，并且能够利用HTTP/2的多路复用特性。
FastAPI后端实现示例：
在FastAPI中，我们可以使用 StreamingResponse 结合异步生成器来实现SSE。
# main.py (FastAPI application)
from fastapi import FastAPI, Response
from fastapi.responses import StreamingResponse
import asyncio
import json
import time
app = FastAPI()
# 模拟硬件状态
hardware_status = {"temperature": 25, "pressure": 1000, "online": True}
# 模拟硬件状态变化的函数
async def simulate_hardware_updates():
while True:
# 假设硬件状态每隔一段时间可能变化
await asyncio.sleep(5) # 每5秒检查一次
new_temperature = hardware_status["temperature"] + (1 if time.time() % 2 == 0 else -1)
if new_temperature < 20: new_temperature = 20
if new_temperature > 30: new_temperature = 30
if new_temperature != hardware_status["temperature"]:
hardware_status["temperature"] = new_temperature
print(f"Hardware status changed: {hardware_status}")
yield f"data: {json.dumps(hardware_status)}\n\n"
else:
# 如果状态没变，可以不发送数据，或者发送一个心跳包
yield "event: heartbeat\ndata: {}\n\n"
@app.get("/hardware-status-sse")
async def sse_hardware_status():
"""
通过SSE推送硬件状态更新。
"""
return StreamingResponse(
simulate_hardware_updates(),
media_type="text/event-stream"
)
# 可以在后台运行一个任务来真正更新 hardware_status
# 例如，通过一个全局变量或消息队列React前端实现示例：
前端通过 EventSource API 订阅SSE流。
// HardwareStatusDisplay.jsx (React Component)
import React, { useState, useEffect } from 'react';
function HardwareStatusDisplay() {
const [status, setStatus] = useState({});
const [isConnected, setIsConnected] = useState(false);
useEffect(() => {
// 创建EventSource实例，连接到FastAPI的SSE端点
const eventSource = new EventSource('http://localhost:8000/hardware-status-sse');
eventSource.onopen = () => {
console.log('SSE connection opened.');
setIsConnected(true);
};
// 监听 'message' 事件，这是默认的事件类型
eventSource.onmessage = (event) => {
console.log('Received SSE message:', event.data);
try {
const newStatus = JSON.parse(event.data);
setStatus(newStatus);
} catch (error) {
console.error('Failed to parse SSE data:', error);
}
};
// 监听自定义事件，例如 'heartbeat'
eventSource.addEventListener('heartbeat', (event) => {
console.log('Received heartbeat:', event.data);
});
eventSource.onerror = (error) => {
console.error('SSE Error:', error);
setIsConnected(false);
eventSource.close(); // 发生错误时关闭连接
};
// 组件卸载时关闭EventSource连接
return () => {
eventSource.close();
console.log('SSE connection closed.');
};
}, []); // 空数组表示只在组件挂载和卸载时运行
return (
<div>
<h2>硬件状态实时监控 (SSE)</h2>
<p>连接状态: {isConnected ? '已连接' : '已断开'}</p>
{Object.keys(status).length > 0 ? (
<ul>
{Object.entries(status).map(([key, value]) => (
<li key={key}>
<strong>{key}:</strong> {String(value)}
</li>
))}
</ul>
) : (
<p>等待硬件状态数据...</p>
)}
</div>
);
}
export default HardwareStatusDisplay;WebSocket 是一种在单个TCP连接上进行全双工通信的协议。与SSE的单向性不同，WebSocket允许客户端和服务器之间进行双向、实时的数据交换。
适用场景： WebSocket适用于需要客户端和服务器频繁双向通信的场景，如在线聊天、多人游戏、实时协作文档编辑等。它提供了更低的延迟和更高的效率，但相对于SSE来说，实现和管理也更为复杂。
FastAPI后端实现示例：
FastAPI内置了对WebSocket的良好支持。
# main.py (FastAPI application - 添加 WebSocket 部分)
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
import asyncio
import json
import time
# ... (上面的 FastAPI app 和 hardware_status 定义不变) ...
# WebSocket连接管理器
class ConnectionManager:
def __init__(self):
self.active_connections: list[WebSocket] = []
async def connect(self, websocket: WebSocket):
await websocket.accept()
self.active_connections.append(websocket)
def disconnect(self, websocket: WebSocket):
self.active_connections.remove(websocket)
async def send_personal_message(self, message: str, websocket: WebSocket):
await websocket.send_text(message)
async def broadcast(self, message: str):
for connection in self.active_connections:
await connection.send_text(message)
manager = ConnectionManager()
# 模拟硬件状态变化的函数 (用于WebSocket)
async def hardware_status_broadcaster():
while True:
await asyncio.sleep(5) # 每5秒检查一次
new_temperature = hardware_status["temperature"] + (1 if time.time() % 2 == 0 else -1)
if new_temperature < 20: new_temperature = 20
if new_temperature > 30: new_temperature = 30
if new_temperature != hardware_status["temperature"]:
hardware_status["temperature"] = new_temperature
print(f"Hardware status changed (WS): {hardware_status}")
await manager.broadcast(json.dumps(hardware_status))
# WebSocket通常不需要心跳，因为连接本身是持久的
@app.websocket("/ws/hardware-status")
async def websocket_endpoint(websocket: WebSocket):
await manager.connect(websocket)
try:
# 第一次连接时发送当前状态
await websocket.send_text(json.dumps(hardware_status))
# 保持连接活跃，等待客户端消息（如果需要）
while True:
data = await websocket.receive_text()
print(f"Received message from client: {data}")
# 如果客户端发送消息，可以根据消息进行处理
except WebSocketDisconnect:
manager.disconnect(websocket)
print("Client disconnected from WebSocket.")
# 启动一个后台任务来持续广播硬件状态
@app.on_event("startup")
async def startup_event():
asyncio.create_task(hardware_status_broadcaster())React前端实现示例：
前端使用浏览器原生的 WebSocket API。
// HardwareStatusWebSocketDisplay.jsx (React Component)
import React, { useState, useEffect, useRef } from 'react';
function HardwareStatusWebSocketDisplay() {
const [status, setStatus] = useState({});
const [isConnected, setIsConnected] = useState(false);
const ws = useRef(null); // 使用ref来保存WebSocket实例
useEffect(() => {
// 创建WebSocket实例
ws.current = new WebSocket('ws://localhost:8000/ws/hardware-status');
ws.current.onopen = () => {
console.log('WebSocket connection opened.');
setIsConnected(true);
// 连接成功后可以发送一些初始化消息给服务器
// ws.current.send(JSON.stringify({ type: 'init', clientId: 'react-app' }));
};
ws.current.onmessage = (event) => {
console.log('Received WebSocket message:', event.data);
try {
const newStatus = JSON.parse(event.data);
setStatus(newStatus);
} catch (error) {
console.error('Failed to parse WebSocket data:', error);
}
};
ws.current.onclose = () => {
console.log('WebSocket connection closed.');
setIsConnected(false);
// 可以尝试重新连接
};
ws.current.onerror = (error) => {
console.error('WebSocket Error:', error);
setIsConnected(false);
// ws.current.close(); // 发生错误时关闭连接
};
// 组件卸载时关闭WebSocket连接
return () => {
if (ws.current) {
ws.current.close();
console.log('WebSocket connection closed on unmount.');
}
};
}, []);
// 示例：如果需要从前端发送数据到后端
const sendMessage = () => {
if (ws.current && ws.current.readyState === WebSocket.OPEN) {
ws.current.send(JSON.stringify({ action: 'request_full_status' }));
} else {
console.warn('WebSocket not connected.');
}
};
return (
<div>
<h2>硬件状态实时监控 (WebSocket)</h2>
<p>连接状态: {isConnected ? '已连接' : '已断开'}</p>
{Object.keys(status).length > 0 ? (
<ul>
{Object.entries(status).map(([key, value]) => (
<li key={key}>
<strong>{key}:</strong> {String(value)}
</li>
))}
</ul>
) : (
<p>等待硬件状态数据...</p>
)}
{/* <button onClick={sendMessage} disabled={!isConnected}>发送消息到后端</button> */}
</div>
);
}
export default HardwareStatusWebSocketDisplay;在决定使用SSE还是WebSocket时，需要考虑以下几点：
数据流向：
实现复杂度：
连接开销：
代理和防火墙：
总结： 根据原始问题描述，硬件状态变化可能长时间不发生，且主要是后端向前端推送数据。在这种情况下，Server-Sent Events (SSE) 是一个非常合适的选择。它提供了轻量级的服务器到客户端推送机制，且具有自动重连的特性，非常适合处理不频繁但需要实时通知的事件。如果未来需求演变为前端也需要频繁地向后端发送指令或消息，那么再考虑升级到WebSocket会更合适。
通过采用SSE或WebSocket，我们可以有效地将FastAPI后端与React前端连接起来，实现高效、实时的硬件状态更新，从而极大地提升应用的响应性和用户体验，同时避免了传统轮询带来的性能瓶颈。
以上就是FastAPI与React实时通信：实现后端主动推送硬件状态更新的详细内容，更多请关注php中文网其它相关文章！
每个人都需要一台速度更快、更稳定的 PC。随着时间的推移，垃圾文件、旧注册表数据和不必要的后台进程会占用资源并降低性能。幸运的是，许多工具可以让 Windows 保持平稳运行。
Copyright 2014-2025 https://www.php.cn/ All Rights Reserved | php.cn | 湘ICP备2023035733号
589
收藏
