FastAPI如何主动向React前端推送硬件状态变更数据？

针对你提到的前端轮询效率低的问题，以下是几种适合的实时推送方案，结合FastAPI和React实现：

1. WebSocket 双向实时通信 #

FastAPI原生支持WebSocket，适合需要双向交互的场景（比如前端偶尔主动查询状态）。

后端实现（FastAPI） #

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import List
import json

app = FastAPI()

# 维护当前连接的客户端列表
active_connections: List[WebSocket] = []

async def broadcast_state(state: dict):
    """向所有在线客户端推送最新状态"""
    for conn in active_connections:
        await conn.send_json(state)

@app.websocket("/ws/hardware-state")
async def websocket_handler(websocket: WebSocket):
    await websocket.accept()
    active_connections.append(websocket)
    try:
        # 新连接建立时先推送当前状态
        current_state = {"status": "normal", "temperature": 25}  # 实际从检测脚本获取
        await websocket.send_json(current_state)
        # 保持连接，可处理前端主动请求（不需要可省略循环）
        while True:
            data = await websocket.receive_text()
            if data == "refresh":
                await websocket.send_json(current_state)
    except WebSocketDisconnect:
        active_connections.remove(websocket)

# 硬件检测脚本更新状态时调用此接口触发推送
@app.post("/update-hardware-state")
async def update_state(state: dict):
    await broadcast_state(state)
    return {"msg": "状态已推送至所有客户端"}

前端实现（React） #

import { useEffect, useState } from 'react';

const HardwareStatus = () => {
  const [hardwareState, setHardwareState] = useState({});

  useEffect(() => {
    const ws = new WebSocket('ws://localhost:8000/ws/hardware-state');

    ws.onopen = () => console.log('WebSocket连接成功');
    ws.onmessage = (e) => setHardwareState(JSON.parse(e.data));
    ws.onclose = () => {
      console.log('WebSocket断开，3秒后重连');
      setTimeout(() => window.location.reload(), 3000);
    };

    return () => ws.close();
  }, []);

  return (
    <div className="status-container">
      <h3>硬件实时状态</h3>
      <p>运行状态：{hardwareState.status || '未获取'}</p>
      <p>当前温度：{hardwareState.temperature || '--'}°C</p>
    </div>
  );
};

export default HardwareStatus;

2. Server-Sent Events (SSE) 单向推送 #

适合仅后端向前端推送数据的场景，协议轻量，无需维护双向连接，资源消耗更低。

后端实现（FastAPI） #

from fastapi import FastAPI, Response
from fastapi.responses import StreamingResponse
import asyncio
import json

app = FastAPI()

# 存储当前硬件状态（实际由检测脚本更新）
current_hardware_state = {"status": "normal", "temperature": 25}

async def state_stream() -> str:
    while True:
        # SSE格式要求：data: JSON字符串\n\n
        yield f"data: {json.dumps(current_hardware_state)}\n\n"
        # 这里可改为监听状态变更信号，而非定时轮询
        await asyncio.sleep(0.5)

@app.get("/sse/hardware-state")
async def sse_handler():
    return StreamingResponse(state_stream(), media_type="text/event-stream")

# 检测脚本调用此接口更新状态
@app.post("/update-hardware-state")
async def update_state(state: dict):
    global current_hardware_state
    current_hardware_state = state
    return {"msg": "状态已更新"}

前端实现（React） #

import { useEffect, useState } from 'react';

const HardwareStatus = () => {
  const [hardwareState, setHardwareState] = useState({});

  useEffect(() => {
    const eventSource = new EventSource('http://localhost:8000/sse/hardware-state');

    eventSource.onmessage = (e) => setHardwareState(JSON.parse(e.data));
    eventSource.onerror = () => {
      console.log('SSE连接异常');
      eventSource.close();
    };

    return () => eventSource.close();
  }, []);

  return (
    <div className="status-container">
      <h3>硬件实时状态</h3>
      <p>运行状态：{hardwareState.status || '未获取'}</p>
      <p>当前温度：{hardwareState.temperature || '--'}°C</p>
    </div>
  );
};

export default HardwareStatus;

3. Redis 发布订阅（跨服务状态同步） #

如果硬件检测脚本和FastAPI是独立部署的服务，用Redis做中间件解耦状态更新与推送逻辑。

步骤1：硬件检测脚本发布状态到Redis #

import redis
import json
import time

# 连接Redis
r = redis.Redis(host='localhost', port=6379, db=0)

# 模拟定期检测硬件状态
while True:
    # 替换为实际检测逻辑
    new_state = {"status": "normal", "temperature": 25 + (time.time() % 5)}
    # 发布到Redis频道
    r.publish('hardware_state_topic', json.dumps(new_state))
    time.sleep(5)

步骤2：FastAPI订阅Redis频道并推送前端 #

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
import redis
import json
import asyncio

app = FastAPI()
r = redis.Redis(host='localhost', port=6379, db=0)
pubsub = r.pubsub()
pubsub.subscribe('hardware_state_topic')

active_connections = []

async def redis_listener():
    """监听Redis频道，收到状态更新则推送给客户端"""
    while True:
        msg = pubsub.get_message(ignore_subscribe_messages=True)
        if msg:
            state = json.loads(msg['data'])
            await broadcast_state(state)
        await asyncio.sleep(0.1)

async def broadcast_state(state: dict):
    for conn in active_connections:
        await conn.send_json(state)

# 启动时开启Redis监听任务
@app.on_event("startup")
async def startup():
    asyncio.create_task(redis_listener())

@app.websocket("/ws/hardware-state")
async def websocket_handler(websocket: WebSocket):
    await websocket.accept()
    active_connections.append(websocket)
    try:
        while True:
            await websocket.receive_text()
    except WebSocketDisconnect:
        active_connections.remove(websocket)

方案选择建议 #

• 需双向交互（前端主动查询/控制）：选WebSocket
• 仅后端单向推送：选SSE，实现简单、资源占用低
• 检测脚本与FastAPI独立部署：用Redis Pub/Sub解耦服务

内容的提问来源于stack exchange，提问作者maja95