做无人机行业应用开发的兄弟十有八九会遇到这么一个问题设备端、机场端、云端之间到底用什么协议传指令和状态比较稳有人用HTTP轮询有人用WebSocket但如果你接的是大疆上云API绕不开的其实是MQTT。我最初接触“大疆无人机 MQTT消息定义”这个主题时以为就是把几个topic订阅好、收收JSON就完事了真正排查起来才发现topic怎么拼、消息体里的method怎么对齐、云端和设备端的时序怎么对上全是细节。这篇就结合我实际接入大疆机场和行业无人机的经验把MQTT消息定义这件事从原理到实操讲透给后面接上云API、做无人机管理平台的兄弟们一个能直接参考的底稿。先说清楚这篇适合谁。如果你正在做无人机机巢、无人机机场、云端管控平台或者想把大疆设备的状态、任务、媒体信息接到自己的后台那这篇就是给你写的。如果你只是拿遥控器飞一下不涉及二次开发可以关掉了。另外文章里涉及具体的topic和消息体格式我会以目前公开的上云API约定为准大疆偶尔会调整字段以官方最新文档为最终依据。1. 先搞清楚大疆无人机的MQTT到底管什么1.1 大疆上云API里MQTT的位置大疆针对行业应用的接入方案核心是上云API它把无人机、机场、云端平台、移动端App串成一张网。这里面的通信链路其实是分层的视频流走RTMP或者GB28181媒体文件通过对象存储和预签名URL来传而设备指令、状态上报、告警事件、航线任务下发这类的“控制面和监控面”消息走的就是MQTT。我习惯把MQTT在这套体系里的作用理解成“消息总线”。飞机起飞、降落、返航、报警、电池电量变化、任务执行进度这些消息全部通过MQTT从机场或飞机侧推到云端云端要下发航线、控制返航、设置机场参数也是把指令发布到对应的topic上由机场侧订阅并执行。整个过程是异步的设备不用等云端慢慢处理云端也不用一直挂着一个长连接去等设备。刚入门的兄弟最容易犯的一个错误是把MQTT当成“实时数传通道”来用想着能不能拿它传图传画面或者高频传感器数据。这个方向基本不对。MQTT在大疆体系里的定位是低频率、高实时性、强可靠性的信令通道真正的大流量数据视频、照片、日志走的是它自己的专用通道。你只需要把MQTT理解成“飞机和云端说话的对讲机”而且是那种只说关键事、不说废话的对讲机。1.2 为什么不用HTTP轮询而用MQTT我做第一个无人机管理后台的时候最开始图省事让云端每5秒去调一次设备接口拉状态结果一接入真实机场就崩了。原因很直接HTTP请求是单向的、同步的云端必须主动去问设备“你状态怎么样”设备才有机会回答。如果设备数量少还好一旦机场数量上了几十台轮询频率一高云端压力大设备侧的API也被频繁调用网络稍微一抖动状态就断层。MQTT天然是双向、长连接的。云端和设备端都连接到同一个Broker设备状态变化时主动往topic上推云端实时订阅就能收到云端要下发指令往另一个topic上发设备端也实时能拿到。这样不用轮询也不会漏消息而且长连接在弱网环境下的表现比短连接好得多自动重连机制也省了很多事。再一个关键点是消息的“推拉关系”。HTTP轮询本质上是你去拿消息但设备端很多事件是偶发的比如“机场舱盖异常打开”“飞机降落成功”“电量低于20%”这类事件如果靠轮询去抓很难设置一个合适的间隔间隔长了不实时间隔短了浪费资源。MQTT的发布订阅模型让设备端在事件发生时主动推送云端不用关心设备什么时候会产生消息只需要订阅固定topic等着收就行。1.3 哪些数据走MQTT哪些不走这块我做接入方案评审时经常要跟客户对齐因为很多人以为“MQTT这么万能把所有数据都塞进去不就行了”。真不行。大疆的架构里数据通道是分流的你对接的时候一定要把这几个通道分清楚。MQTT负责的核心数据类型有这些设备上下线生命周期事件device_online、device_offline、OSD遥测数据经纬度、高度、速度、电量这类、飞行任务的下发与进度、机场设备状态、告警事件、媒体文件上传完成的通知。这些消息的特征是单条消息体量小、实时性要求高、对时序有强依赖。不走MQTT的数据最典型的是视频流。视频流延时要低、带宽要大MQTT的报文头虽然小但拿它承载连续媒体流根本不现实。媒体文件如拍照照片、录像文件也是同理设备完成上传后只是通过MQTT推一个“媒体文件上传完成”的事件里面带上文件的存储路径、对象存储地址、元数据真正的文件内容由云端去对象存储里拉取或者通过预签名URL直接下载。把这层逻辑梳理清楚之后你再去设计自己的后端模块就知道该监听哪些topic、该把哪些字段落库、哪些数据该走另外的通道。如果一上来就把所有消息混在一个地方处理后期排查问题会非常痛苦。2. 大疆MQTT消息定义的底层逻辑2.1 Topic结构设计一眼看懂你这包消息要去哪大疆MQTT的topic定义是整个消息体系里最值得先吃透的部分因为所有消息流转都是靠topic来路由的。官方采用的是一套“$thing/上行还是下行/product/设备型号/消息类型”的分层结构熟悉物联网平台的同学看到这个会很眼熟这跟市面上主流IoT平台如阿里云IoT、腾讯云IoT的主题设计思路基本一致只不过前缀和设备标识规则换成了大疆自己的。实际接入中你主要会跟这几类topic打交道Topic前缀方向作用$thing/down/product/{product_id}/services云端到设备云端下发服务调用比如创建航线任务、控制返航$thing/up/product/{product_id}/services设备到云端服务调用的返回结果对应下行服务的响应$thing/up/product/{product_id}/events设备到云端设备主动上报的事件比如设备上下线、告警、媒体通知$thing/up/product/{product_id}/property设备到云端设备属性上报物模型里定义的属性值$thing/down/product/{product_id}/property云端到设备云端设置设备属性部分版本开放$thing/up/product/{product_id}/lifecycle设备到云端设备生命周期消息核心是设备上线、离线通知这里的{product_id}不是设备SN而是“产品ID”也就是你在大疆开发者平台创建产品时拿到的那一串标识。同一个产品ID下面会挂多台物理设备所以单靠product_id定位不到具体某台飞机还要在消息体或者clientId里带上设备标识。这个设计跟我们平时做IoT平台时的产品Product、设备Device两层模型是一致的。关于topic大小写我踩过一次坑。大疆这套topic是全小写的services、events、property这些单词没有大写有些兄弟习惯性地写成Services或EventsBroker对topic的匹配是大小写敏感的订阅成功后却一直收不到消息查了半天才发现是这个原因。另外通配符也可以灵活用订阅端想一次性收到所有产品的服务返回可以用$thing/up///services这里的代表一层通配#则代表多层通配。但生产环境我一般不建议用太宽泛的通配符宁可多写几个订阅精确到product_id避免串消息。2.2 消息体的统一骨架与关键字段topic解决的是“消息去哪”的问题消息体解决的是“消息里到底说了什么”的问题。大疆的MQTT消息体统一采用JSON格式而且不管是事件上报还是服务下发都套了一个固定的外框。这个设计很实用接入方只需要解析一次外层结构就能知道这条消息的类型、请求目的和时序关系。典型的消息体长这样{ tid: uuid-xxx-xxx, bid: uuid-yyy-yyy, timestamp: 1710000000000, method: flight_task_create, data: { flight_id: xxxx, task_type: 0, waylines: [] } }这里的四个顶层字段非常重要。tid是消息的事务ID一般由消息发起方生成用来唯一标识一次交互云端下发服务时生成一个tid设备端在处理完成后返回响应时会把同一个tid带回这样云端就能把“请求”和“响应”对上号bid是业务ID可以理解成你给这次业务操作起的一个业务层面的标识比如某次航线任务的业务编号timestamp是毫秒级时间戳用于消息时序判断和日志排查method是消息的方法名也就是“这条消息要做什么事”比如flight_task_create表示创建飞行任务device_online表示设备上线osd_info表示OSD遥测数据上报。data字段里面放的是具体的业务参数不同method对应不同的data结构。我建议在代码里做一层“协议映射”把method字符串和对应的数据解析类绑定在一起解析的时候先用method路由再反序列化data这样即便后面新增了方法也只是加一个分支不会把主逻辑搅乱。有一点要特别注意tid和bid都不能为空而且同一设备这边尽量复用同一个bid来做一次完整业务流程比如整个航线任务从下发到结束都用同一个bid排查问题时按bid去日志里拉链路非常方便。2.3 鉴权与安全三元组加HMAC签名到底怎么算大疆MQTT接入跟AWS IoT、阿里云IoT的接入方式很像不是随便拿一个客户端配上地址就能连的。你需要先准备好“三元组”也就是Product ID、Device SN、Device Secret。Product ID是产品的唯一标识Device SN是物理设备的序列号Device Secret是设备密钥这三样东西在开发者平台创建产品和注册设备时就能拿到。MQTT连接时Client ID、Username、Password这三项都不是随便填的。大疆用的规则一般是Client ID:{product_id}_{device_sn}Username:{product_id}_{device_sn}Password: 时间戳加Device Secret做HMAC-SHA256签名后的结果具体签名串怎么拼我贴一段我实际跑通的Python代码这个你拿过去改改三元组就能用import hmac import hashlib import time def generate_password(device_secret: str) - str: # 使用毫秒级时间戳作为签名内容 timestamp str(int(time.time() * 1000)) message timestamp.encode(utf-8) secret device_secret.encode(utf-8) sign hmac.new(secret, message, hashlib.sha256).hexdigest() return sign注意几个细节。第一时间戳一定要和设备侧当前时间保持同步误差太大会导致签名校验失败我遇到过因为服务器时区没核对、导致密码一直报错的情况后来统一用UTC毫秒时间戳解决。第二HMAC的key是device_secretmessage是时间戳字符串这个顺序不能反反了签名结果就完全不同。第三有些固件版本或者接口模式下MQTT连接会和设备注册流程绑定如果你用的是大疆官方的零代码方案或者已经接入过大疆云端的一键体验方案鉴权方式可能上云API已经封装好了不需要自己写签名。连接地址和端口方面通常MQTT走TLS加密连接端口一般是8883或者443具体的Broker域名在上云API文档里能找到。企业网络环境要注意放通对应端口不然客户端一直显示连接超时。如果做本地联调有些开发模式也支持走非加密端口但生产环境务必用TLS因为MQTT报文的控制指令可能涉及飞行安全明文传输风险很大。2.4 QoS到底选0、1还是2MQTT的QoSQuality of Service是很多人容易忽略的配置项但在大疆场景里选错了会很麻烦。QoS0是最多一次消息发出去就不管了可能丢QoS1是至少一次保证送达但可能重复QoS2是恰好一次最严格但开销最大交互握手也最重。在大疆的MQTT消息链路里我实测下来推荐这样配链路方向推荐QoS原因云端下发服务指令QoS1指令必须到达设备端QoS0不可接受设备上行事件上报QoS1状态事件尽量不丢重复可以靠tid去重OSD高频遥测数据QoS0高频数据丢一两帧没影响QoS1反而增加Broker压力为什么连接认证、设备上下线这类关键消息不开QoS2因为QoS2的协议开销大而且对Broker的会话状态有要求实际大疆Broker未必对QoS2做了完整支持我在生产环境里统一用QoS1业务层对重复消息做幂等处理。具体做法收到一条消息先按tid查一下最近处理过的记录如果存在就跳过。这个方法比依赖QoS2更灵活大部分IoT平台也是这么干的。OSD那种一秒一次甚至更快的高频遥测用QoS0完全没问题网络断了丢几帧无所谓重连后最新状态会继续推上来。3. 实操记录把一台大疆设备接入MQTT全流程3.1 准备阶段搭一个能看包的MQTT调试环境正式写业务代码之前我强烈建议先把MQTT调试环境搭起来否则你都不知道收到的那包数据长什么样后面写解析代码全靠猜。常用工具有MQTT X、MQTT Explorer和开源的paho-mqtt库。MQTT X适合图形化地订阅和发布消息MQTT Explorer适合看topic结构而写代码调试就用paho-mqtt。如果你手头没有真实的大疆机场设备也不要紧。大疆的上云API平台一般提供模拟器或者沙箱环境开发者可以在云端创建一个虚拟设备模拟器会连上MQTT Broker并且自动上报设备上线、OSD等消息。我实际用下来模拟器对验证topic结构、消息格式非常有帮助唯一的区别是模拟器的事件节奏比真实设备规整很多真实设备偶尔会冒出一些异常事件字段但大框架一致。另外如果你需要自己搭一个Broker做联调用Docker跑EMQX是最快的一条命令就能起一个带管理界面的Broker方便看连接数、订阅关系和消息流向。但要注意自建Broker只能是联通性联调真正接入大疆平台还是要连大疆提供的Broker因为消息路由规则在它们那边。docker run -d --name emqx -p 1883:1883 -p 18083:18083 emqx/emqx:latest跑起来之后浏览器访问http://localhost:18083就能进入EMQX Dashboard默认用户名admin密码public。这个环境对大疆MQTT消息定义的前期学习特别有帮助你可以拿它反复实验订阅主题、发布消息看看消息是怎么路由的。3.2 连接建立与设备上线事件解析我直接贴一段paho-mqtt的Python代码这是我自己在用的最小化接入模板你替换掉三元组和连接地址就能跑通连接和上线事件订阅import json import hmac import hashlib import time import paho.mqtt.client as mqtt PRODUCT_ID your_product_id DEVICE_SN your_device_sn DEVICE_SECRET your_device_secret BROKER_HOST your_broker_host BROKER_PORT 8883 def generate_password(device_secret: str) - str: timestamp str(int(time.time() * 1000)) message timestamp.encode(utf-8) secret device_secret.encode(utf-8) return hmac.new(secret, message, hashlib.sha256).hexdigest() client_id f{PRODUCT_ID}_{DEVICE_SN} username f{PRODUCT_ID}_{DEVICE_SN} client mqtt.Client(client_idclient_id, protocolmqtt.MQTTv311) client.tls_set() client.username_pw_set(username, generate_password(DEVICE_SECRET)) def on_connect(client, userdata, flags, rc): print(connect result:, rc) # 订阅设备生命周期事件注意topic里的通配符 client.subscribe(f$thing/up/product/{PRODUCT_ID}/lifecycle, qos1) client.subscribe(f$thing/up/product/{PRODUCT_ID}/events, qos1) def on_message(client, userdata, msg): payload json.loads(msg.payload.decode(utf-8)) print(ftopic: {msg.topic}, method: {payload.get(method)}) if payload.get(method) device_online: print(设备上线:, payload.get(data)) if payload.get(method) device_offline: print(设备离线:, payload.get(data)) client.on_connect on_connect client.on_message on_message client.connect(BROKER_HOST, BROKER_PORT, keepalive60) client.loop_forever()这段代码跑通之后你会在控制台看到device_online事件不断打出来。这里有个关键点client_id和username都要用product_id_device_sn的格式拼password用签名函数生成如果你在别的语言里实现注意拼写和签名算法保持一致。收到device_online事件时data里一般会带设备的经纬度、安装信息、设备型号等基础信息。这个事件非常关键很多业务逻辑都以设备上线为起点比如设备在线后自动同步当前状态、刷新机场列表、检查固件版本等。我习惯在device_online处理逻辑里去触发一次属性同步请求主动拉取机场当前所有物模型属性的值。3.3 下发飞行任务从云端到飞机的完整消息流接入MQTT后最核心的业务就是下发飞行任务。大疆的上云API里创建航线任务对应的方法名是flight_task_create它的topic是$thing/down/product/{product_id}/services。云端往这个topic发布消息设备端收到后执行执行结果再通过$thing/up/product/{product_id}/services返回。我贴一个创建航线任务的下发报文这样看得更直观{ tid: e5a7f8b1-xxxx-4c72-9a20-xxxxxxxxxx, bid: task-20240601-001, timestamp: 1710000000000, method: flight_task_create, data: { flight_id: task-20240601-001, task_type: 0, wayline_file_url: https://your-oss-bucket.oss-cn-xxxx.aliyuncs.com/wayline.wpml, execute_mode: 0, speed: 8, rtk_switch: 1 } }这个报文发出去后设备端会很快返回一个响应响应消息的tid和请求的tid一致method一般是flight_task_create_reply或者带reply后缀的形式data里带一个result字段标识指令接收情况。注意这里只代表“指令被设备接收了”并不代表“任务真正执行成功”。任务后续的执行进度会通过flight_task_progress这样的事件以独立的消息推送到events或services的topic上。整个消息流是这样的下发创建任务请求→ 设备返回接收结果响应→ 推任务进度后续事件→ 任务结束结束事件。你在后端设计时需要把这三类消息通过bid关联起来形成一个完整的任务生命周期。我踩过的坑是只处理了“下发”和“接收结果”没处理后续的进度事件结果前端一直看不到任务执行过程后来才把进度事件加上。另外任务下发之前一定要确认设备在线否则消息会发布到Broker上但设备收不到。虽然MQTT有持久会话可以帮设备缓存离线消息但大疆的命令消息很多是即时性的过期就没有意义了。我在代码里会做一个在线状态校验设备不在线直接拒绝下发提示前端“设备离线无法执行任务”。3.4 状态回传与OSD消息解析设备飞行过程中最核心的遥测数据就是OSD消息。OSDOn-Screen Display在大疆的语境里是指飞机实时状态叠加数据包含经纬度、海拔、高度、水平速度、垂直速度、飞行模式、电池电量、遥控器信号、图传信号等。这些数据会上报到$thing/up/product/{product_id}/events主题method为osd_info。我见过不少兄弟一上来就把所有OSD字段全部落库结果数据库表爆炸。实际使用中应该按业务需要做字段裁剪比如只需要定位和电量那就只解析longitude、latitude、height、battery这几个字段。OSD消息的频率不低我自己接的机场OSD消息大约一秒到几秒一条高峰时一天几万条很常见所以入库前一定要加一个“状态变化判断”或者“抽样存储”的策略。做前端大屏展示的时候OSD消息可以经过WebSocket直接推给浏览器但不要直接把MQTT原始报文转发出去因为里面有大量无关字段而且数据格式对前端不友好。我通常的做法是后端订阅OSD消息解析出需要字段重新组装成一个精简的实时状态对象再通过WebSocket推给前端。这样前端只关注视图刷新不用理解大疆的协议。4. 常见问题与排查技巧实录4.1 连接失败先查签名、再做网络连通性检查MQTT连接不上是接入时最常见的故障。我排查的顺序是先看connect回调里的返回码返回码0是连接成功非0的对应关系在MQTT协议里有明确说明如果返回码对不上优先查三元组和签名。签名错的情况占大头。常见错误是时间戳用秒而不是毫秒或者签名用的message内容跟平台端期望不一致。我建议在本地先把签名结果打印出来再拿同一个时间戳放到平台的在线调试工具里比对看是否一致。如果签名能对上那就要查端口防火墙、TLS证书配置。有些Broker对TLS版本有要求Python的tls_set()默认会用系统证书如果你的环境内网有代理或者公司网关做了SSL解密连接也会失败这种情况要在运维侧把Broker域名加入白名单。还有一种隐蔽情况多个进程或者多台服务器用同一个{product_id}_{device_sn}作为clientId去连接导致后一个连接把前一个踢下线。这个在MQTT协议里叫“会话接管”表现就是设备端上报正常但云端连接时不时掉线。排查时看一下是不是有多处代码在裸连统一收口到一个连接池服务里管理。4.2 订阅不到消息八成是topic拼写或层级问题能连上Broker但收不到消息这个问题排查起来也很考验耐心。首先确认你订阅的topic和发布端发布的topic完全一致这里的一致包括大小写、下划线位置、product_id是否写错。{product_id}一旦出错所有订阅都会变成“订阅了空气”。其次是通配符问题。$thing/up///events是合法订阅但如果你把写成了*那是无效的订阅会被静默拒绝。再就是$开头的topic在MQTT协议里属于特殊主题有些Broker在权限配置里会对$开头topic做限制如果你发现订阅$thing/...总是不行检查一下Broker端的订阅权限和ACL规则。如果topic检查了没问题就要打开抓包或日志看看设备端是不是真的发送了消息。用MQTT Explorer或者EMQX Dashboard都可以看到当前topic下的消息流情况。有些时候不是你没订阅到而是设备端根本没有上报比如设备还在启动中或者设备绑定关系没建立好。我遇到过一次很奇怪的场景设备状态都正常但OSD消息就是不来后来发现是设备端固件版本太低不支持新版的osd_info协议升级固件后正常了。4.3 消息发出去没反应先确认设备在线和method是否匹配云端往$thing/down/product/{product_id}/services发了指令但设备没执行这个问题的排查路径一般是三步。第一步确认设备在线设备不在线一切免谈。第二步确认method和data结构是不是设备端支持的版本。大疆不同品类的设备支持的method集合有差异比如机场的支持flight_task_create但某些机型不一定支持同样的字段。第三步查看设备端有没有返回error响应很多情况下设备是收到了消息但因为参数不满足条件比如机场舱盖没关闭、RTK未定位任务创建会被拒绝并返回带错误码的响应。我处理过最经典的一个案例是下发返航指令设备端一直没反应查了半天发现data里需要带home_latitude和home_longitude参数而我只传了指令method没有带返航点坐标。设备端校验不通过直接把消息丢弃了。所以看消息定义的时候要认真看方法对应的必填字段不能想当然。4.4 时序问题先处理上线再处理业务消息MQTT的异步特性决定了消息到达顺序和发送顺序不一定完全一致。比如云端同时下发“创建任务”和“立即返航”两条指令设备端可能先处理了后收到的返航指令。这种时序问题在无人机场景里很危险轻则任务混乱重则产生误操作。我的应对方案是在业务层做状态机和指令队列。所有下发到设备的指令先进入本地队列按bid做分组同一组指令按顺序发送设备端返回响应之前不发送下一条同类型的指令。同时收到设备端的状态事件后先更新本地的设备状态机再处理具体业务逻辑。比如只有设备状态为“空闲”的时候才允许下发航线任务设备状态为“返航中”时禁止下发起飞指令。另外一个容易被忽略的细节是设备上下线事件的到达顺序。设备可能刚上报device_online马上又上报device_offline如果你在业务里把离线当成设备故障去告警就会产生误报。我后来加了一个防抖窗口收到device_offline事件后先等10秒如果在10秒内又有device_online事件就说明是网络闪断或者设备重启不触发告警。这样处理之后误报率明显下降。4.5 高频消息的幂等处理大疆MQTT虽然QoS一般用1但网络重连、Broker重投机制都可能导致消息重复送达。我之前在事件处理逻辑里没有做幂等结果重复的媒体文件上传通知导致后端重复创建了三份媒体记录数据库里全是脏数据。处理办法很简单维护一个最近消息ID的去重表。每次收到消息先看tid或bid是否已经处理过如果处理过就丢弃。去重表可以用Redis键名用mqtt:dedup:{tid}设置过期时间比如5分钟。对于高频的OSD消息还可以直接用时间戳加设备标识做去重键减少Redis写入压力。我个人在实际操作中的体会是大疆这套MQTT消息定义的边界非常清晰只要你把topic、消息体、会话和时序这四件事想清楚了整个接入过程就顺了。最怕的就是边写边猜发现收不到消息再回头查文档。最后再分享一个小技巧联调阶段把原始报文完整打印出来包括topic和payload保留三天日志出了问题按bid一搜就能定位这是我排查线上问题效率最高的方式。