ESP32 MQTT对接巴法云平台

ESP32 MQTT对接巴法云平台

MQTT（Message Queuing Telemetry Transport）是一种轻量级的 发布/订阅（Publish/Subscribe） 消息传输协议，专为 低带宽、高延迟、不稳定网络环境 设计，是物联网（IoT）领域的核心通信协议之一。
MQTT 核心特性

特性说明轻量级协议头最小仅 2字节，适合嵌入式设备发布/订阅模型设备不直接通信，通过 Broker（代理服务器） 中转，解耦性强低功耗适合电池供电设备，支持 心跳包（Keep Alive） 维持长连接QoS 质量等级支持 0/1/2 三级消息可靠性保证主题（Topic）分层式消息路由（如 home/living_room/temperature），支持通配符 + 和 #MQTT 与 HTTP 对比

对比维度MQTTHTTP协议开销极低（适合高频小数据）高（Header 冗余）通信模式双向实时推送单向请求-响应连接开销长连接（减少握手延迟）短连接（每次请求需重建连接）适用场景物联网、移动推送、实时监控Web 服务、API 接口MQTT 核心概念

• Broker（代理服务器）
  
  
  
  • 核心枢纽，负责消息路由（如 Mosquitto、EMQX、HiveMQ）
    
  • 实现消息存储转发、客户端管理、安全认证
    
  
  
• Topic（主题）
  
  
  
  • 消息分类的层级路径（例如：factory/machine1/status）
    
  • 通配符：
    
    
    
    • +：单级匹配（home/+/temperature 匹配 home/living_room/temperature）
      
    • #：多级匹配（home/# 匹配 home/living_room/light）
      
    
    
  
  
• QoS（服务质量）
  
  
  
  • QoS 0：最多一次（尽力交付，可能丢失）
    
  • QoS 1：至少一次（确保送达，可能重复）
    
  • QoS 2：恰好一次（严格保证，无重复）
    
  
  

micro python可行性先行验证

1. # 导入必要的库
2. from umqtt.simple import MQTTClient  # MQTT协议客户端库
3. import time                          # 时间相关函数
4. from machine import Timer            # 硬件定时器控制
6. #################### 用户可修改配置区域 ####################
7. wifiName = "jianzhiji02"             # WiFi名称（仅支持2.4G网络）
8. wifiPassword = "8765432111"          # WiFi密码
9. clientID = ""  # 设备密钥，从巴法云控制台获取
10. subTopic = 'OTALX'                   # 订阅的主题（接收指令）
11. pubTopic = 'bafa'               # 发布的主题（发送数据）
13. #################### 固定配置区域 ########################
14. serverIP = "bemfa.com"               # MQTT服务器地址
15. port = 9501                          # MQTT端口号
16. ping_interval = 300                  # 心跳包间隔（秒）
18. ################# WiFi连接函数 ##########################
19. def do_connect():
20.     """连接WiFi网络"""
21.     import network
22.     sta_if = network.WLAN(network.STA_IF)
23.     if not sta_if.isconnected():
24.         print('正在连接网络...')
25.         sta_if.active(True)
26.         sta_if.connect(wifiName, wifiPassword)
27.         while not sta_if.isconnected():
28.             pass
29.     print('WiFi连接成功')
30.     print('网络配置:', sta_if.ifconfig())
32. ################# MQTT消息回调函数 #######################
33. def msg_callback(topic, msg):
34.     """处理接收到的MQTT消息"""
35.     global client  # 声明全局客户端对象
36.    
37.     print("[消息到达] 主题:", topic.decode(), "| 载荷:", msg.decode())
38.    
39.     # 判断是否是订阅的主题
40.     if topic.decode() == subTopic:
41.         # 根据指令执行操作
42.         if msg == b"on":
43.             print("执行开机操作")
44.             # 示例：控制GPIO输出高电平
45.             # pin.value(1)
46.             
47.             # 发送状态确认（推送消息）
48.             client.publish(pubTopic, "设备已开启")
49.             
50.         elif msg == b"off":
51.             print("执行关机操作")
52.             # 示例：控制GPIO输出低电平
53.             # pin.value(0)
54.             
55.             # 发送状态确认（推送消息）
56.             client.publish(pubTopic, "设备已关闭")
58. ################ MQTT连接与订阅函数 #####################
59. def connect_mqtt():
60.     """建立MQTT连接并订阅主题"""
61.     global client  # 声明全局客户端对象
62.    
63.     try:
64.         # 创建客户端实例
65.         client = MQTTClient(clientID, serverIP, port)
66.         client.set_callback(msg_callback)  # 设置回调函数
67.         
68.         # 建立连接
69.         client.connect()
70.         print(f"成功连接到MQTT服务器 {serverIP}:{port}")
71.         
72.         # 订阅主题
73.         client.subscribe(subTopic)
74.         print(f"已订阅主题: {subTopic}")
75.         
76.         # 设置心跳间隔（可选）
77.         client.keepalive = ping_interval
78.         
79.         return client
80.    
81.     except Exception as e:
82.         print("MQTT连接失败:", e)
83.         restart_and_reconnect()
85. ################ 异常处理函数 ##########################
86. def restart_and_reconnect():
87.     """重启设备并重连"""
88.     print("10秒后重启设备...")
89.     time.sleep(10)
90.     machine.reset()
92. ################ 定时推送函数 ##########################
93. def timed_publish(timer):
94.     """定时发布设备状态（示例）"""
95.     global client
96.     try:
97.         # 示例数据（可替换为传感器数据）
98.         status = "11"
99.         client.publish(pubTopic, f" {status}")
100.         print(f"定时推送: {status}")
101.     except:
102.         print("定时推送失败")
104. ################# 主程序流程 ###########################
105. if __name__ == "__main__":
106.     # 连接WiFi
107.     do_connect()
108.    
109.     # 初始化MQTT连接
110.     connect_mqtt()
111.    
112.     # 设置定时器（每30秒发送心跳）
113.     timer = Timer(-1)  # 创建虚拟定时器
114.     timer.init(
115.         period=3000,    # 间隔3秒（3000毫秒）
116.         mode=Timer.PERIODIC,
117.         callback=timed_publish
118.     )
119.     print("定时推送已启用")
120.    
121.     # 主循环
122.     try:
123.         while True:
124.             try:
125.                 client.check_msg()  # 检查新消息
126.                 time.sleep(1)       # 降低CPU占用
127.                
128.             except Exception as e:
129.                 print("运行错误:", e)
130.                 restart_and_reconnect()
131.                
132.     except KeyboardInterrupt:
133.         print("程序终止")
134.         client.disconnect()
135.         timer.deinit()

复制代码

终端打印

成功连接云平台，成功推送，订阅主题。可行性验证成功。
云平台

巴法云物联网平台_MQTT设备云
推送内容控制led

接受ESP32推送内容

Esp-idf+C语言实现

重要代码段分析

MQTT推送

1. //----------------MQTT推送-----------------------------------------------------------------//
2. //推送字符
3.         const char *status = "putting";
4.         //mqtt句柄，推送目标主题，推送内容，长度，服务质量，发布消息的标志位（通常设置为0）
5.         esp_mqtt_client_publish(mqtt_client, PUB_TOPIC, status, strlen(status), 0, 0);
6.   //推送数字
7.         num+=3;//推送运行时间
8.         snprintf(num_str, sizeof(num_str), "%d", num); // 将整数转为字符串
9.         esp_mqtt_client_publish(mqtt_client, PUB_TOPIC2, num_str, strlen(num_str), 0, 0);

复制代码

MQTT推送函数

esp_mqtt_client_publish：这是ESP-IDF提供的函数，用于通过MQTT客户端发布消息。

• mqtt_client：这是MQTT客户端的句柄，通常是在初始化MQTT客户端时创建的。
  
• PUB_TOPIC：这是您要发布消息的主题名称，通常是一个字符串。
  
• status：这是您要发布的消息内容，通常是一个字符串。
  
• strlen(status)：这是消息内容的长度，strlen函数用于计算字符串的长度。
  ：这是QoS（Quality of Service）等级，表示服务质量。MQTT协议定义了三种QoS等级：
  
  
  
  • 0：最多一次（"fire-and-forget"）
    
  • 1：至少一次
    
  • 2：只有一次
    
  
  
• 0：这是发布消息的标志位，通常设置为0。
  

MQTT订阅（MQTT成功连接）

1. // MQTT事件处理
2. static void mqtt_event_handler(void *handler_args, esp_event_base_t base,
3.                               int32_t event_id, void *event_data) {
4.     esp_mqtt_event_handle_t event = event_data;
6.     switch (event->event_id) {
7.         case MQTT_EVENT_CONNECTED://<MQTT连接事件>---------->MQTT成功连接则订阅主题
8.             ESP_LOGI(TAG, "MQTT Connected");
9.             mqtt_connected = true;
10.             //----------------MQTT订阅(MQTT成功后)-----------------------------------------
11.             //mqtt句柄,订阅的主题，服务等级（0）
12.             esp_mqtt_client_subscribe(mqtt_client, SUB_TOPIC, 0);//订阅主题SUB_TOPIC（ledctrl）
13.             
14.             
15.         case MQTT_EVENT_DATA: {//<MQTT接收数据事件>
16.             // 处理接收数据
17.             //----------------判断接收主题-----------------------------------------------------------------//
18.             char topic[event->topic_len + 1];
19.             memcpy(topic, event->topic, event->topic_len);
20.             topic[event->topic_len] = '\0';
21.             //----------------判断接收数据-----------------------------------------------------------------//
22.             char data[event->data_len + 1];
23.             memcpy(data, event->data, event->data_len);
24.             data[event->data_len] = '\0';
25.             //----------------打印接收主题+数据-----------------------------------------------------------------//
26.             ESP_LOGI(TAG, "Received: Topic=%s, Data=%s", topic, data);
28.             if (strcmp(topic, SUB_TOPIC) == 0) {//判断接收主题(ledctrl)数据——>控制led灯->推送led状态信息
29.                 if (strcmp(data, "on") == 0) {
30.                     gpio_set_level(OUTPUT_PIN, 0);
31.                     //推送led状态信息到主题ledstate
32.                     esp_mqtt_client_publish(mqtt_client, PUB_TOPIC3, "led设备已开启", 0, 0, 0);//同时推送状态到主题ledstate
33.                 } else if (strcmp(data, "off") == 0) {
34.                     gpio_set_level(OUTPUT_PIN, 1);
35.                     esp_mqtt_client_publish(mqtt_client, PUB_TOPIC3, "led设备已关闭", 0, 0, 0);
36.                 }
37.             }
38.             break;
39.         }
40.         default:
41.             break;
42.     }
43. }

复制代码

运行逻辑：

MQTT成功连接（MQTT事件）->订阅主题
MQTT接收数据事件->判断接收内容

MQTT订阅函数

esp_mqtt_client_subscribe(mqtt_client, SUB_TOPIC, 0); // 订阅主题SUB_TOPIC（ledctrl）

• esp_mqtt_client_subscribe：这是ESP-IDF提供的函数，用于订阅MQTT主题。
  
• mqtt_client：这是MQTT客户端的句柄，通常是在初始化MQTT客户端时创建的。
  
• SUB_TOPIC：这是您要订阅的主题名称，通常是一个字符串。在这个例子中，主题名称是ledctrl，表示用于控制LED灯的主题。
  
• 0：这是QoS（Quality of Service）等级，表示服务质量。MQTT协议定义了三种QoS等级：
  
  
  
  • 0：最多一次（"fire-and-forget"）
    
  • 1：至少一次
    
  • 2：只有一次
    
  
  

终端现象分析

成功连接WIFI

成功连接MQTT服务器，并推送主题+内容

接受到MQTT服务器订阅的主题内容

完整代码：

1. #include <string.h>
2. #include "freertos/FreeRTOS.h"
3. #include "freertos/task.h"
4. #include "esp_system.h"
5. #include "esp_log.h"
6. #include "esp_event.h"
7. #include "esp_netif.h"
8. #include "nvs_flash.h"
9. #include "protocol_examples_common.h"
10. #include "mqtt_client.h"
11. #include "driver/gpio.h"
12. #include "esp_timer.h"
13. #include "esp_wifi.h"
14. // 用户配置参数
15. #define WIFI_SSID      "jianzhiji02"//wifi名称
16. #define WIFI_PASS      "8765432111"//wifi密码
17. #define MQTT_CLIENT_ID ""//巴法云平台秘钥
18. #define SUB_TOPIC      "ledctrl"//订阅主题（控制核心板led）
19. #define PUB_TOPIC      "bafa"//推送主题1（测试推送字符）
20. #define PUB_TOPIC2     "Time"//推送主题2（测试推送数字）
21. #define PUB_TOPIC3     "ledstate"//推送主题2（推送led状态）
24. // #define BROKER_URI     "mqtt://bemfa.com:9501"//服务器URL（需DNS解析）
25. #define BROKER_URI     "mqtt://119.91.109.180:9501"//服务器ip地址+端口
26. static const char *TAG = "MQTT_DEMO";
27. static esp_mqtt_client_handle_t mqtt_client = NULL;//mqtt句柄
28. static esp_timer_handle_t status_timer;//状态定时器句柄
29. static bool mqtt_connected = false;//mqtt连接状态
31. #define OUTPUT_PIN     GPIO_NUM_48//led 引脚（低电平点亮）
32. // GPIO初始化
33. static void init_gpio(void) {
34.     gpio_config_t io_conf = {
35.         .pin_bit_mask = (1ULL << OUTPUT_PIN),
36.         .mode = GPIO_MODE_OUTPUT,
37.         .pull_up_en = GPIO_PULLUP_DISABLE,
38.         .pull_down_en = GPIO_PULLDOWN_DISABLE,
39.         .intr_type = GPIO_INTR_DISABLE
40.     };
41.     gpio_config(&io_conf);
42.     gpio_set_level(OUTPUT_PIN, 1);
43. }
46. // 定义静态变量 // 缓冲区用于存储转换后的字符串
47. static int num = 0;static char num_str[16];
48. // 定时器回调函数（1s周期）
49. //----------------MQTT推送-----------------------------------------------------------------//
50. static void publish_status(void *arg) {
51.     if (mqtt_connected) {
52.         const char *status = "putting";//推送主题
53.         esp_mqtt_client_publish(mqtt_client, PUB_TOPIC, status, strlen(status), 0, 0);
54.         ESP_LOGI(TAG, "Published status: %s", status);
55.    
56.         num+=3;//推送运行时间
57.         snprintf(num_str, sizeof(num_str), "%d", num); // 将整数转为字符串
58.         esp_mqtt_client_publish(mqtt_client, PUB_TOPIC2, num_str, strlen(num_str), 0, 0);
59.         ESP_LOGI(TAG, "num: %d", num);  // 正确打印整型
60.     }
61. }
63. // MQTT事件处理
64. static void mqtt_event_handler(void *handler_args, esp_event_base_t base,
65.                               int32_t event_id, void *event_data) {
66.     esp_mqtt_event_handle_t event = event_data;
68.     switch (event->event_id) {
69.         case MQTT_EVENT_CONNECTED://<MQTT连接事件>
70.             ESP_LOGI(TAG, "MQTT Connected");
71.             mqtt_connected = true;
72.             //----------------MQTT订阅-----------------------------------------------------------------//
73.             esp_mqtt_client_subscribe(mqtt_client, SUB_TOPIC, 0);//订阅主题ledctrl
74.             // 创建状态定时器（3秒周期）
75.             esp_timer_create_args_t timer_cfg = {
76.                 .callback = &publish_status,
77.                 .name = "status_timer"
78.             };
79.             esp_timer_create(&timer_cfg, &status_timer);
80.             esp_timer_start_periodic(status_timer, 3000000);
81.             break;
83.         case MQTT_EVENT_DISCONNECTED:
84.             ESP_LOGI(TAG, "MQTT Disconnected");
85.             mqtt_connected = false;
86.             esp_timer_stop(status_timer);
87.             break;
89.         case MQTT_EVENT_DATA: {//<MQTT接收数据事件>
90.             // 处理接收数据
91.             //----------------判断接收主题-----------------------------------------------------------------//
92.             char topic[event->topic_len + 1];
93.             memcpy(topic, event->topic, event->topic_len);
94.             topic[event->topic_len] = '\0';
95.             //----------------判断接收数据-----------------------------------------------------------------//
96.             char data[event->data_len + 1];
97.             memcpy(data, event->data, event->data_len);
98.             data[event->data_len] = '\0';
99.             //----------------打印接收主题+数据-----------------------------------------------------------------//
100.             ESP_LOGI(TAG, "Received: Topic=%s, Data=%s", topic, data);
102.             if (strcmp(topic, SUB_TOPIC) == 0) {//判断接收主题(ledctrl)数据——>控制led灯->推送led状态信息
103.                 if (strcmp(data, "on") == 0) {
104.                     gpio_set_level(OUTPUT_PIN, 0);
105.                     //推送led状态信息到主题ledstate
106.                     esp_mqtt_client_publish(mqtt_client, PUB_TOPIC3, "led设备已开启", 0, 0, 0);//同时推送状态到主题ledstate
107.                 } else if (strcmp(data, "off") == 0) {
108.                     gpio_set_level(OUTPUT_PIN, 1);
109.                     esp_mqtt_client_publish(mqtt_client, PUB_TOPIC3, "led设备已关闭", 0, 0, 0);
110.                 }
111.             }
112.             break;
113.         }
115.         case MQTT_EVENT_ERROR:
116.             ESP_LOGE(TAG, "MQTT Error");
117.             break;
119.         default:
120.             break;
121.     }
122. }
124. // WiFi事件处理
125. static void wifi_event_handler(void *arg, esp_event_base_t event_base,
126.                               int32_t event_id, void *event_data) {
127.     if (event_id == IP_EVENT_STA_GOT_IP) {
128.         // WiFi连接成功后启动MQTT
129.         esp_mqtt_client_config_t mqtt_cfg = {
130.             .broker.address.uri = BROKER_URI,
131.             .credentials.client_id = MQTT_CLIENT_ID
132.         };
133.         mqtt_client = esp_mqtt_client_init(&mqtt_cfg);
134.         esp_mqtt_client_register_event(mqtt_client, ESP_EVENT_ANY_ID, mqtt_event_handler, NULL);
135.         esp_mqtt_client_start(mqtt_client);
136.     }
137. }
139. void app_main(void) {
140.     // 初始化NVS（堆栈存储wifi连接数据等。。。）
141.     esp_err_t ret = nvs_flash_init();
142.     if (ret == ESP_ERR_NVS_NO_FREE_PAGES || ret == ESP_ERR_NVS_NEW_VERSION_FOUND) {
143.         ESP_ERROR_CHECK(nvs_flash_erase());
144.         ret = nvs_flash_init();
145.     }
146.     ESP_ERROR_CHECK(ret);
148.     // 初始化网络接口
149.     ESP_ERROR_CHECK(esp_netif_init());
150.     ESP_ERROR_CHECK(esp_event_loop_create_default());
151.    
152.     // 配置WiFi
153.     esp_netif_create_default_wifi_sta();
154.     wifi_init_config_t cfg = WIFI_INIT_CONFIG_DEFAULT();
155.     ESP_ERROR_CHECK(esp_wifi_init(&cfg));
156.    
157.     // 注册WiFi事件
158.     esp_event_handler_instance_t instance_any_id;
159.     ESP_ERROR_CHECK(esp_event_handler_instance_register(IP_EVENT,
160.                                                         IP_EVENT_STA_GOT_IP,
161.                                                         &wifi_event_handler,
162.                                                         NULL,
163.                                                         &instance_any_id));
164.     // 设置WiFi参数
165.     wifi_config_t wifi_config = {
166.         .sta = {
167.             .ssid = WIFI_SSID,
168.             .password = WIFI_PASS,
169.         },
170.     };
171.     ESP_ERROR_CHECK(esp_wifi_set_mode(WIFI_MODE_STA));//设置为STA模式
172.     ESP_ERROR_CHECK(esp_wifi_set_config(WIFI_IF_STA, &wifi_config));//设置WiFi参数
173.     ESP_ERROR_CHECK(esp_wifi_start());//启动WiFi
174.     ESP_ERROR_CHECK(esp_wifi_connect());//连接WiFi
176.     // 初始化GPIO(led)
177.     init_gpio();
179.     // 保持主任务运行
180.     while (1) {
181.         vTaskDelay(pdMS_TO_TICKS(1000));
182.     }
183. }

复制代码

如果文章对你有所帮助，可以帮我点一下左下角推荐该文，万分感谢

博主目前在广州，深圳找实习。
如有大佬（HR,BOSS）能推荐实习，恳请私信小弟，给个机会，帮帮小弟，万分感谢！！！

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！