（七）RT-Thread物联网实战--MQTT-cJSON-OneNET

RT-Thread物联网实战 — MQTT cJSON OneNET云平台前言设备联网之后如何将数据上传到云平台并实现远程监控本文将串联MQTT协议、Paho MQTT库、cJSON数据解析以及OneNET云平台构建完整的物联网数据链路传感器 → MCU → ESP8266 → MQTT → OneNET云平台。一、MQTT 协议1.1 什么是MQTTMQTTMessage Queuing Telemetry Transport是一种发布/订阅模式的轻量级消息传输协议专为低带宽、不稳定网络和低功耗场景设计是物联网通信的首选协议。1.2 核心角色角色说明比喻发布者Publisher发送数据的设备投稿人订阅者Subscriber接收数据的设备读者代理Broker消息路由中心报社编辑部主题Topic消息分类标签报纸栏目发布者和订阅者不直接通信而是通过Broker中转实现松耦合。1.3 工作流程设备连接Broker通过TCP长连接使用唯一的客户端ID注册。发布者发布消息指定主题发送数据如/home/livingroom/temperature→25°C。Broker转发消息找到所有订阅该主题的客户端逐一推送。订阅者接收消息处理接收到的数据。1.4 主题通配符单层通配符/home//temperature匹配所有房间的温度多层通配符#/home/#匹配家中所有传感器数据1.5 QoS 消息质量QoS级别保证适用场景QoS 0至多一次可能丢失温度数据等非关键信息QoS 1至少一次可能重复开关控制指令QoS 2恰好一次不丢不重报警状态等关键信息1.6 保留消息与遗嘱消息保留消息Broker保存主题的最新消息新订阅者加入时立即推送适合状态同步。遗嘱消息客户端连接时预设意外断开时Broker自动发布适合故障通知。二、Paho MQTT 库Paho MQTT 是RT-Thread中常用的MQTT客户端库提供完整的连接、订阅、发布功能。2.1 订阅列表配置MQTTClient client;client.messageHandlers[0].topicFilter/device/data;client.messageHandlers[0].callbackmqtt_sub_callback;client.messageHandlers[0].qosQOS1;2.2 回调函数回调触发时机connect_callbackMQTT连接成功online_callback客户端上线offline_callback客户端掉线defaultMessageHandler默认消息接收messageHandlers[x].callback指定主题消息接收2.3 URI 格式tcp://iot.eclipse.org:1883 // 域名 tcp://192.168.10.1:1883 // IPv4 ssl://192.168.10.1:1884 // SSL加密 tcp://[fe80::20c:29ff:...]:1883 // IPv62.4 核心APIAPI说明paho_mqtt_start(client)启动MQTT客户端paho_mqtt_stop(client)停止MQTT客户端paho_mqtt_subscribe(client, qos, topic, cb)订阅主题paho_mqtt_unsubscribe(client, topic)取消订阅paho_mqtt_publish(client, qos, topic, msg)发布消息paho_mqtt_control(client, cmd, arg)控制客户端参数控制命令命令功能MQTT_CTRL_SET_CONN_TIMEO设置连接超时时间MQTT_CTRL_SET_RECONN_INTERVAL设置重连间隔MQTT_CTRL_SET_KEEPALIVE_INTERVAL设置心跳间隔MQTT_CTRL_PUBLISH_BLOCK设置发布阻塞/非阻塞模式三、cJSON 库cJSON是轻量级JSON解析/生成库物联网中用于构建上传数据和解析下发命令。3.1 核心数据结构typedefstructcJSON{structcJSON*next,*prev;// 链表指针structcJSON*child;// 子节点对象/数组inttype;// 节点类型char*valuestring;// 字符串值intvalueint;// 整数值doublevaluedouble;// 浮点值char*string;// 键名}cJSON;支持的类型cJSON_False、cJSON_True、cJSON_NULL、cJSON_Number、cJSON_String、cJSON_Array、cJSON_Object3.2 解析 JSON/* 解析JSON字符串 */constchar*json_str{\temperature\:25,\humidity\:60};cJSON*rootcJSON_Parse(json_str);/* 获取字段值 */cJSON*tempcJSON_GetObjectItem(root,temperature);if(cJSON_IsNumber(temp)){rt_kprintf(温度: %d\n,temp-valueint);}/* 释放内存 */cJSON_Delete(root);3.3 构建 JSON/* 创建JSON对象 */cJSON*rootcJSON_CreateObject();cJSON_AddNumberToObject(root,temperature,25);cJSON_AddStringToObject(root,status,normal);/* 转为字符串 */char*json_strcJSON_PrintUnformatted(root);rt_kprintf(JSON: %s\n,json_str);// {temperature:25,status:normal}/* 释放 */cJSON_free(json_str);cJSON_Delete(root);3.4 常用API速查操作API解析JSONcJSON_Parse(str)创建对象cJSON_CreateObject()创建数组cJSON_CreateArray()添加数字cJSON_AddNumberToObject(obj, key, num)添加字符串cJSON_AddStringToObject(obj, key, str)添加子对象cJSON_AddItemToObject(obj, key, item)获取字段cJSON_GetObjectItem(obj, key)获取数组元素cJSON_GetArrayItem(arr, index)转为字符串cJSON_Print(obj)/cJSON_PrintUnformatted(obj)释放对象cJSON_Delete(obj)类型检查cJSON_IsNumber()/cJSON_IsString()等3.5 RT-Thread 内存适配cJSON_Hooks hooks;hooks.malloc_fnrt_malloc;hooks.free_fnrt_free;cJSON_InitHooks(hooks);四、OneNET 云平台OneNET是中国移动推出的物联网开放平台支持设备接入、数据存储、可视化展示和远程控制。4.1 工作原理OneNET软件包基于MQTT和HTTP Client实现数据上传通过MQTT向特定topic发布消息命令下发Broker将服务器命令推送给设备数据查询通过HTTP GET/POST请求获取历史数据┌─────────┐ MQTT ┌──────────┐ HTTP ┌──────────┐ │ 设备 │ ─────────→ │ OneNET │ ←─────────→ │ 网页/APP │ │ (MCU) │ ←───────── │ Broker │ │ (用户) │ └─────────┘ 命令下发 └──────────┘ 数据查询 └──────────┘4.2 依赖库库用途cJSON构建/解析JSON数据Paho MQTTMQTT连接、发布、订阅HTTP Client数据查询、设备注册4.3 初始化intonenet_mqtt_init(void);内部流程设置MQTT参数URI、设备ID、用户名、密码注册回调函数连接/上线/掉线/消息接收启动MQTT客户端/* 核心初始化逻辑 */mq_client.uritcp://183.230.40.96:1883;mq_client.condata.clientID.cstringdevice_id;mq_client.condata.username.cstringproduct_id;mq_client.condata.password.cstringauth_info;mq_client.defaultMessageHandlermqtt_callback;paho_mqtt_start(mq_client);4.4 数据上传API上传数字数据rt_err_tonenet_mqtt_upload_digit(constchar*ds_name,constdoubledigit);/* 上传温度数据到 temperature 数据流 */onenet_mqtt_upload_digit(temperature,25.5);上传字符串数据rt_err_tonenet_mqtt_upload_string(constchar*ds_name,constchar*str);/* 上传设备状态 */onenet_mqtt_upload_string(status,running);上传二进制数据rt_err_tonenet_mqtt_upload_bin(constchar*ds_name,constuint8_t*bin,size_tlen);从文件路径上传rt_err_tonenet_mqtt_upload_bin_by_path(constchar*ds_name,constchar*bin_path);向指定主题发布rt_err_tonenet_mqtt_publish(constchar*topic,constuint8_t*msg,size_tlen);4.5 命令接收与响应设置命令响应回调当云端下发命令时自动调用voidcmd_response(uint8_t*recv_data,size_trecv_size,uint8_t**resp_data,size_t*resp_size){rt_kprintf(收到命令: %.*s\n,recv_size,recv_data);/* 构建响应 */*resp_data(uint8_t*)OK;*resp_size2;}/* 注册回调 */onenet_set_cmd_rsp_cb(cmd_response);4.6 数据查询APIHTTPAPI说明onenet_http_get_datastream(name, info)获取数据流信息onenet_get_dp_by_limit(name, n)获取最近N个数据点onenet_get_dp_by_start_end(name, start, end, n)获取指定时间段数据点4.7 设备管理APIAPI说明onenet_http_register_device(name, auth)注册新设备onenet_port_save_device_info(id, key)保存设备信息需用户实现onenet_port_get_register_info(name, auth)获取注册信息需用户实现onenet_port_get_device_info(id, key, auth)获取设备信息需用户实现onenet_port_is_registed()查询设备是否已注册五、完整物联网数据链路将前面所有知识串联起来一个完整的物联网项目数据流如下┌─────────┐ GPIO ┌──────────┐ UARTAT ┌──────────┐ │ DHT11 │ ──────────→ │ STM32 │ ──────────→ │ ESP8266 │ │ 传感器 │ 温湿度数据 │ RT-Thread│ AT命令控制 │ WiFi模块 │ └─────────┘ └──────────┘ └──────────┘ │ │ cJSON构建数据 MQTT发布 │ │ ▼ ▼ ┌──────────┐ ┌──────────┐ │ OneNET │ ←────────── │ MQTT │ │ 云平台 │ 数据入库 │ Broker │ └──────────┘ └──────────┘ │ 网页/APP展示关键技术栈对应环节技术RT-Thread组件数据采集GPIO/传感器框架PIN设备、传感器框架设备联网AT命令 WiFiAT组件、ESP8266库数据格式化JSONcJSON库数据传输MQTT协议Paho MQTT库云端对接OneNET APIOneNET软件包总结要点内容MQTT协议发布/订阅模式轻量高效支持QoSPaho MQTTRT-Thread的MQTT客户端库cJSON轻量级JSON解析/生成物联网数据格式标准OneNET中国移动物联网平台支持MQTTHTTP数据链路传感器 → MCU → WiFi → MQTT → 云平台 → 用户