如何实现Python服务器、MQTT与Web客户端响应式通信并解决阻塞问题？

核心问题是你原有的WebSocket处理循环阻塞了事件循环，导致MQTT消息无法及时处理。下面是基于FastAPI、FastMQTT和异步WebSocket的无阻塞实现方案，同时支持双向通信：

关键改进点 #

• 所有IO操作（MQTT收发、WebSocket通信）使用异步方法，让事件循环可以并行处理任务
• 用异步锁保护活跃WebSocket连接集合，避免并发修改问题
• MQTT消息回调直接异步广播给所有在线Web客户端，无需轮询

完整代码实现 #

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastmqtt import FastMQTT, MQTTConfig
import asyncio
import json
from pydantic import BaseModel
from datetime import datetime

# 定义MQTT消息格式模型（验证用）
class MqttImageMessage(BaseModel):
    source: str
    measure: str  # Base64编码的图片字符串
    timestamp: datetime

# 初始化FastAPI实例
app = FastAPI()

# MQTT broker配置（替换成你的Mosquitto信息）
mqtt_config = MQTTConfig(
    host="localhost",
    port=1883,
    # 如果需要认证，取消下面两行注释
    # username="mqtt_user",
    # password="mqtt_pass"
)
mqtt = FastMQTT(config=mqtt_config)

# 维护活跃WebSocket连接集合+异步锁（避免并发修改）
active_ws_connections = set()
connection_lock = asyncio.Lock()

# 启动/关闭MQTT客户端
@app.on_event("startup")
async def startup_event():
    await mqtt.connection()

@app.on_event("shutdown")
async def shutdown_event():
    await mqtt.disconnect()

# MQTT消息处理回调（异步）
@mqtt.subscribe("image/stream/#")  # 替换成你的MQTT主题
async def handle_mqtt_incoming(client, topic, payload, qos, properties):
    try:
        # 解析并验证MQTT消息
        msg_data = json.loads(payload.decode())
        validated_msg = MqttImageMessage(**msg_data)
        
        # 广播消息给所有活跃WebSocket客户端
        async with connection_lock:
            # 遍历集合副本，避免修改时出错
            for ws in list(active_ws_connections):
                try:
                    await ws.send_json(validated_msg.dict())
                except WebSocketDisconnect:
                    active_ws_connections.remove(ws)
                except Exception as e:
                    print(f"Failed to send to WS: {str(e)}")
                    active_ws_connections.remove(ws)
    except Exception as e:
        print(f"Invalid MQTT message: {str(e)}")

# WebSocket端点（双向通信）
@app.websocket("/ws/image-stream")
async def ws_image_stream(websocket: WebSocket):
    await websocket.accept()
    # 将当前连接加入活跃集合
    async with connection_lock:
        active_ws_connections.add(websocket)
    
    try:
        # 异步接收客户端消息（非阻塞，事件循环可处理其他任务）
        async for client_msg in websocket.iter_text():
            # 处理客户端消息，比如转发到MQTT
            await mqtt.publish("client/commands", client_msg)
            # 或者自定义业务逻辑
            print(f"Received from client: {client_msg}")
    except WebSocketDisconnect:
        pass
    finally:
        # 连接关闭后移除集合
        async with connection_lock:
            active_ws_connections.discard(websocket)

为什么这个方案能解决阻塞问题？ #

1. 异步IO模型：所有网络操作（MQTT连接、WebSocket收发）都是异步的，事件循环在等待IO时会切换到其他任务（比如处理MQTT消息或其他WebSocket连接）
2. 无阻塞的WebSocket循环：用async for遍历客户端消息，这是异步迭代器，不会阻塞事件循环
3. 并发安全的连接管理：用asyncio.Lock保护连接集合，避免多任务同时修改导致的异常

可选优化/替代方案 #

• Redis Pub/Sub扩展：如果需要横向扩展多个FastAPI实例，可将MQTT消息转发到Redis频道，WebSocket客户端订阅Redis频道，实现跨实例消息广播
• Python 3.11+ TaskGroup：用asyncio.TaskGroup管理WebSocket的接收和消息转发任务，代码更简洁
• 消息分片：如果Base64图片过大，可将消息分片发送，避免WebSocket消息大小限制

注意事项 #

• 调整FastAPI的websocket_max_size参数，支持大尺寸Base64图片
• 给MQTT设置合适的QoS级别（比如QoS=1），确保图片消息不丢失
• 定期清理无效的WebSocket连接，避免内存泄漏

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