扫描二维码
关注或者微信搜一搜：编程智域 前端至全栈交流与成长

发现1000+提升效率与开发的AI工具和实用程序：https://tools.cmdragon.cn/

MQTT协议基础与FastAPI集成原理
MQTT（Message Queuing Telemetry Transport）是一种基于发布/订阅模式的轻量级消息协议，采用"主题-消息"
机制实现设备间通信。其工作流程可通过以下时序图表示：

[设备A] --发布消息到/temperature主题--> [MQTT Broker]
[FastAPI服务] --订阅/temperature主题--> [MQTT Broker]
[Broker] --推送消息--> [FastAPI服务]

# 导入必要的库
from fastapi import FastAPI
from fastapi_mqtt import FastMQTT
from pydantic import BaseModel
import logging

# 配置日志记录
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("MQTT_OTA")

# 初始化FastAPI应用
app = FastAPI(title="IoT设备OTA升级服务")

# 配置MQTT连接参数（示例使用公共测试服务器）
mqtt_config = {
    "host": "test.mosquitto.org",
    "port": 1883,
    "keepalive": 60,
    "client_id": "fastapi_ota_server"
}

# 初始化MQTT客户端
mqtt = FastMQTT(config=mqtt_config)
mqtt.init_app(app)


# 定义Pydantic数据模型
class FirmwareUpdate(BaseModel):
    device_id: str
    firmware_version: str
    chunk_size: int = 512  # 默认分片大小512KB
    checksum: str


# MQTT连接状态回调
@mqtt.on_connect()
def handle_connect(client, flags, rc, properties):
    logger.info(f"Connected with result code {rc}")
    mqtt.client.subscribe("/ota/+/request")  # 订阅设备升级请求主题


# 设备升级请求处理
@app.post("/firmware/upgrade")
async def create_upgrade_task(update: FirmwareUpdate):
    """创建固件升级任务"""
    # 此处添加数据库记录存储逻辑
    return {"task_id": "OTA_20230801_001", "status": "queued"}


# MQTT消息处理
@mqtt.on_message()
async def message_handler(client, topic, payload, qos, properties):
    """处理设备端MQTT消息"""
    device_id = topic.split("/")[2]

    if "request" in topic:
        handle_upgrade_request(device_id, payload)
    elif "progress" in topic:
        logger.info(f"设备{device_id}升级进度: {payload.decode()}")
    elif "verify" in topic:
        handle_verification(device_id, payload)


def handle_upgrade_request(device_id: str, payload: bytes):
    """处理升级请求"""
    try:
        request_data = json.loads(payload)
        # 验证设备合法性
        if validate_device(device_id):
            # 推送升级元数据
            metadata = {
                "url": f"https://firmware.example.com/{request_data['version']}.bin",
                "size": 2048000,
                "checksum": "sha256:9f86d08..."
            }
            mqtt.publish(f"/ota/{device_id}/metadata", json.dumps(metadata))
    except Exception as e:
        logger.error(f"处理请求错误: {str(e)}")


# 运行参数配置
if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=8000)

代码说明：

1. 依赖库及版本：

   • fastapi==0.103.1
   • fastapi-mqtt==0.1.5
   • pydantic==1.10.7
   • uvicorn==0.23.2

2. 功能实现要点：

   • 使用on_connect装饰器实现MQTT连接成功后的自动订阅
   • 采用主题通配符/ota/+/request监听所有设备的升级请求
   • 通过Pydantic模型严格校验升级请求参数
   • 实现分片传输进度跟踪和校验验证机制
   • 使用异步处理提升高并发下的性能表现

3. 协议升级流程：