ESP32學(xué)習(xí)筆記（46）——MQTT客戶端

Jenny_Tong 發(fā)布于2021-09-09 09:33 / 2265人閱讀

摘要：一簡介實(shí)現(xiàn)方式實(shí)現(xiàn)協(xié)議需要客戶端和服務(wù)器端通訊完成，在通訊過程中，協(xié)議中有三種身份發(fā)布者代理服務(wù)器訂閱者。其中，消息的發(fā)布者和訂閱者都是客戶端，消息代理是服務(wù)器，消息發(fā)布者可以同時(shí)是訂閱者。

一、MQTT簡介

1.1 實(shí)現(xiàn)方式

實(shí)現(xiàn)MQTT協(xié)議需要客戶端和服務(wù)器端通訊完成，在通訊過程中，MQTT協(xié)議中有三種身份：發(fā)布者（Publish）、代理（Broker）（服務(wù)器）、訂閱者（Subscribe）。其中，消息的發(fā)布者和訂閱者都是客戶端，消息代理是服務(wù)器，消息發(fā)布者可以同時(shí)是訂閱者。

MQTT傳輸?shù)南⒎譃椋褐黝}（Topic）和負(fù)載（payload）兩部分：

Topic，可以理解為消息的類型，訂閱者訂閱（Subscribe）后，就會收到該主題的消息內(nèi)容（payload）；

payload，可以理解為消息的內(nèi)容，是指訂閱者具體要使用的內(nèi)容。

MQTT服務(wù)器的主要工作是數(shù)據(jù)分發(fā)，沒有數(shù)據(jù)保存功能。
可以訂閱自己發(fā)布的主題，服務(wù)器就是回發(fā)測試。
MQTT讓邏輯變得更清晰，需要什么訂閱什么。
走標(biāo)準(zhǔn)化流程，解放了私有協(xié)議制定、實(shí)現(xiàn)、調(diào)試、測試一整套復(fù)雜的流程。

1.2 ESP-MQTT

ESP-MQTT 是 MQTT 協(xié)議客戶端的實(shí)現(xiàn)（MQTT 是輕量級的發(fā)布/訂閱消息協(xié)議）。

支持 MQTT over TCP、SSL with mbedtls、MQTT over Websocket、MQTT over Websocket Secure。
使用 URI 輕松設(shè)置
多個(gè)實(shí)例（一個(gè)應(yīng)用程序中有多個(gè)客戶端）
支持訂閱、發(fā)布、身份驗(yàn)證、最后遺囑消息、保持活動 ping 和所有 3 個(gè) QoS 級別（它應(yīng)該是一個(gè)功能齊全的客戶端）。

ESP-IDF 編程指南——ESP-MQTT

二、API說明

以下 MQTT 客戶端接口位于 components/mqtt/esp-mqtt/include/mqtt_client.h 。

2.1 esp_mqtt_client_init

2.2 esp_mqtt_client_register_event

2.3 esp_mqtt_client_start

2.4 esp_mqtt_client_publish

2.5 esp_mqtt_client_subscribe

2.6 esp_mqtt_client_unsubscribe

三、MQTT客戶端

3.1 主要流程

3.2 配置MQTT參數(shù)

首先，要定義一個(gè) MQTT 客戶端配置結(jié)構(gòu)體，最小配置即填入 MQTT 服務(wù)器的 URL 即可。

esp_mqtt_client_config_t mqtt_cfg = {    .uri = CONFIG_BROKER_URL,};

esp_mqtt_client_config_t 結(jié)構(gòu)體如下：

typedef struct {    mqtt_event_callback_t event_handle; /*回調(diào)*/    const char *host; /*!< MQTT 服務(wù)器域名(ipv4 as string)*/    const char *uri; /*!< MQTT 服務(wù)器域名 */    uint32_t port; /*!< MQTT服務(wù)器端口*/    const char *client_id; /*MQTT Client的名字默認(rèn)是ESP32_加上MAC后3hex*/    const char *username; /*MQTT用戶名*/    const char *password; /*MQTT密碼*/    const char *lwt_topic; /*!< LWT主題，默認(rèn)為空*/    const char *lwt_msg; /*!< LWT信息，默認(rèn)為空*/    int lwt_qos; /*!< LWT消息質(zhì)量*/    int lwt_retain; /*!< LWT保留消息標(biāo)志*/    int lwt_msg_len; /*!< LWT消息長度*/    int disable_clean_session; /*!< mqtt clean session,默認(rèn)為真*/    int keepalive; /*MQTT心跳，默認(rèn)120秒 */    bool disable_auto_reconnect; /*錯(cuò)誤，斷開后重連，true不連*/    void *user_context; /*用戶信息 */    int task_prio; /*!< MQTT任務(wù)優(yōu)先級，默認(rèn)為5,可以在make menuconfig中修改*/    int task_stack; /*!< MQTT 任務(wù)堆棧大小,默認(rèn)6144 bytes,可以在make menuconfig中修改*/    int buffer_size; /*!< MQTT收發(fā)緩存，默認(rèn)1024 */    const char *cert_pem; /*指向用于服務(wù)器驗(yàn)證（使用SSL）的PEM格式的證書數(shù)據(jù)的指針，默認(rèn)值為空，不需要驗(yàn)證服務(wù)器 */    const char *client_cert_pem; /*指向用于SSL相互身份驗(yàn)證的PEM格式的證書數(shù)據(jù)的指針，默認(rèn)值為空，如果不需要相互身份驗(yàn)證，則不需要。如果不為空，還必須提供“客戶機(jī)密鑰”。*/    const char *client_key_pem; /*指向用于SSL相互身份驗(yàn)證的PEM格式的私鑰數(shù)據(jù)的指針，默認(rèn)值為空，如果不需要相互身份驗(yàn)證，則不需要。如果不為空，還必須提供“client-cert-pem”。*/    esp_mqtt_transport_t transport; /*覆蓋URI傳輸*/} esp_mqtt_client_config_t;

3.3 初始化MQTT客戶端

然后通過 esp_mqtt_client_init() 獲取一個(gè) MQTT 客戶端結(jié)構(gòu)體指針，參數(shù)是 MQTT 客戶端配置結(jié)構(gòu)體。

esp_mqtt_client_handle_t client = esp_mqtt_client_init(&mqtt_cfg);

3.4 注冊MQTT事件

默認(rèn)情況下，MQTT 客戶端使用事件循環(huán)庫來發(fā)布相關(guān)的 MQTT 事件（已連接，已訂閱，已發(fā)布等）。

所以我們要注冊一個(gè) MQTT 事件，填入 MQTT 事件處理函數(shù) mqtt_event_handler()，

static void mqtt_event_handler(void *handler_args, esp_event_base_t base, int32_t event_id, void *event_data) {    ESP_LOGD(TAG, "Event dispatched from event loop base=%s, event_id=%d", base, event_id);    mqtt_event_handler_cb(event_data);}

第一個(gè)參數(shù)為MQTT客戶端結(jié)構(gòu)體，
第二個(gè)是事件ID對應(yīng)的事件類型，
第三個(gè)參數(shù)即事件處理函數(shù)，
第四個(gè)參數(shù)為事件處理函數(shù)的參數(shù)。

3.5 開啟MQTT客戶端

esp_mqtt_client_start(client);

3.6 MQTT事件處理

static esp_err_t mqtt_event_handler_cb(esp_mqtt_event_handle_t event){    esp_mqtt_client_handle_t client = event->client;    int msg_id;    // your_context_t *context = event->context;    switch (event->event_id) {        case MQTT_EVENT_CONNECTED:            ESP_LOGI(TAG, "MQTT_EVENT_CONNECTED");            msg_id = esp_mqtt_client_publish(client, "/topic/qos1", "data_3", 0, 1, 0);            ESP_LOGI(TAG, "sent publish successful, msg_id=%d", msg_id);            msg_id = esp_mqtt_client_subscribe(client, "/topic/qos0", 0);            ESP_LOGI(TAG, "sent subscribe successful, msg_id=%d", msg_id);            msg_id = esp_mqtt_client_subscribe(client, "/topic/qos1", 1);            ESP_LOGI(TAG, "sent subscribe successful, msg_id=%d", msg_id);            msg_id = esp_mqtt_client_unsubscribe(client, "/topic/qos1");            ESP_LOGI(TAG, "sent unsubscribe successful, msg_id=%d", msg_id);            break;        case MQTT_EVENT_DISCONNECTED:            ESP_LOGI(TAG, "MQTT_EVENT_DISCONNECTED");            break;        case MQTT_EVENT_SUBSCRIBED:            ESP_LOGI(TAG, "MQTT_EVENT_SUBSCRIBED, msg_id=%d", event->msg_id);            msg_id = esp_mqtt_client_publish(client, "/topic/qos0", "data", 0, 0, 0);            ESP_LOGI(TAG, "sent publish successful, msg_id=%d", msg_id);            break;        case MQTT_EVENT_UNSUBSCRIBED:            ESP_LOGI(TAG, "MQTT_EVENT_UNSUBSCRIBED, msg_id=%d", event->msg_id);            break;        case MQTT_EVENT_PUBLISHED:            ESP_LOGI(TAG, "MQTT_EVENT_PUBLISHED, msg_id=%d", event->msg_id);            break;        case MQTT_EVENT_DATA:            ESP_LOGI(TAG, "MQTT_EVENT_DATA");            printf("TOPIC=%.*s/r/n", event->topic_len, event->topic);            printf("DATA=%.*s/r/n", event->data_len, event->data);            break;        case MQTT_EVENT_ERROR:            ESP_LOGI(TAG, "MQTT_EVENT_ERROR");            break;        default:            ESP_LOGI(TAG, "Other event id:%d", event->event_id);            break;    }    return ESP_OK;}

四、示例代碼

根據(jù) examples/protocols/mqtt/tcp 中的例程修改
根據(jù)服務(wù)器地址修改.host = "192.168.61.67",

/* MQTT (over TCP) Example   This example code is in the Public Domain (or CC0 licensed, at your option.)   Unless required by applicable law or agreed to in writing, this   software is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR   CONDITIONS OF ANY KIND, either express or implied.*/#include #include #include #include #include "esp_wifi.h"#include "esp_system.h"#include "nvs_flash.h"#include "esp_event.h"#include "esp_netif.h"#include "protocol_examples_common.h"#include "freertos/FreeRTOS.h"#include "freertos/task.h"#include "freertos/semphr.h"#include "freertos/queue.h"#include "lwip/sockets.h"#include "lwip/dns.h"#include "lwip/netdb.h"#include "esp_log.h"#include "mqtt_client.h"static const char *TAG = "MQTT_EXAMPLE";static esp_err_t mqtt_event_handler_cb(esp_mqtt_event_handle_t event){    // 獲取MQTT客戶端結(jié)構(gòu)體指針    esp_mqtt_client_handle_t client = event->client;    int msg_id;    // your_context_t *context = event->context;    // 通過事件ID來分別處理對應(yīng)的事件    switch (event->event_id) {        case MQTT_EVENT_CONNECTED:    // MQTT連上事件            ESP_LOGI(TAG, "MQTT_EVENT_CONNECTED");            // MQTT Client發(fā)布主題函數(shù)，主題是/topic/qos1，服務(wù)質(zhì)量qos1，發(fā)布的數(shù)據(jù)是data-3            msg_id = esp_mqtt_client_publish(client, "/topic/qos1", "data_3", 0, 1, 0);            ESP_LOGI(TAG, "sent publish successful, msg_id=%d", msg_id);            // MQTT Client訂閱主題函數(shù)，主題是/topic/qos0，服務(wù)質(zhì)量qos0            msg_id = esp_mqtt_client_subscribe(client, "/topic/qos0", 0);            ESP_LOGI(TAG, "sent subscribe successful, msg_id=%d", msg_id);            // MQTT Client訂閱主題函數(shù)，主題是/topic/qos1，服務(wù)質(zhì)量qos1            msg_id = esp_mqtt_client_subscribe(client, "/topic/qos1", 1);            ESP_LOGI(TAG, "sent subscribe successful, msg_id=%d", msg_id);            // MQTT Client取消訂閱主題函數(shù)            msg_id = esp_mqtt_client_unsubscribe(client, "/topic/qos1");            ESP_LOGI(TAG, "sent unsubscribe successful, msg_id=%d", msg_id);            break;        case MQTT_EVENT_DISCONNECTED:    // MQTT斷開連接事件            ESP_LOGI(TAG, "MQTT_EVENT_DISCONNECTED");            break;        case MQTT_EVENT_SUBSCRIBED:    // MQTT發(fā)送訂閱成功事件            ESP_LOGI(TAG, "MQTT_EVENT_SUBSCRIBED, msg_id=%d", event->msg_id);            msg_id = esp_mqtt_client_publish(client, "/topic/qos0", "data", 0, 0, 0);            ESP_LOGI(TAG, "sent publish successful, msg_id=%d", msg_id);            break;        case MQTT_EVENT_UNSUBSCRIBED:    // MQTT取消訂閱事件            ESP_LOGI(TAG, "MQTT_EVENT_UNSUBSCRIBED, msg_id=%d", event->msg_id);            break;        case MQTT_EVENT_PUBLISHED:    // MQTT發(fā)布成功事件            ESP_LOGI(TAG, "MQTT_EVENT_PUBLISHED, msg_id=%d", event->msg_id);            break;        case MQTT_EVENT_DATA:    // MQTT接收數(shù)據(jù)事件            ESP_LOGI(TAG, "MQTT_EVENT_DATA");            printf("TOPIC=%.*s/r/n", event->topic_len, event->topic);            printf("DATA=%.*s/r/n", event->data_len, event->data);            break;        case MQTT_EVENT_ERROR:            ESP_LOGI(TAG, "MQTT_EVENT_ERROR");            break;        default:            ESP_LOGI(TAG, "Other event id:%d", event->event_id);            break;    }    return ESP_OK;}static void mqtt_event_handler(void *handler_args, esp_event_base_t base, int32_t event_id, void *event_data) {    ESP_LOGD(TAG, "Event dispatched from event loop base=%s, event_id=%d", base, event_id);    mqtt_event_handler_cb(event_data);}static void mqtt_app_start(void){     // 1、定義一個(gè)MQTT客戶端配置結(jié)構(gòu)體，輸入MQTT的url    esp_mqtt_client_config_t mqtt_cfg = {        .host = "192.168.61.67",    // MQTT服務(wù)器地址        .port = 1883,               // MQTT服務(wù)器端口    };#if CONFIG_BROKER_URL_FROM_STDIN    char line[128];    if (strcmp(mqtt_cfg.uri, "FROM_STDIN") == 0) {        int count = 0;        printf("Please enter url of mqtt broker/n");        while (count < 128) {            int c = fgetc(stdin);            if (c == "/n") {                line[count] = "/0";                break;            } else if (c > 0 && c < 127) {                line[count] = c;                ++count;            }            vTaskDelay(10 / portTICK_PERIOD_MS);        }        mqtt_cfg.uri = line;        printf("Broker url: %s/n", line);    } else {        ESP_LOGE(TAG, "Configuration mismatch: wrong broker url");        abort();    }#endif /* CONFIG_BROKER_URL_FROM_STDIN */    // 2、通過esp_mqtt_client_init獲取一個(gè)MQTT客戶端結(jié)構(gòu)體指針，參數(shù)是MQTT客戶端配置結(jié)構(gòu)體    esp_mqtt_client_handle_t client = esp_mqtt_client_init(&mqtt_cfg);    // 3、注冊MQTT事件    esp_mqtt_client_register_event(client, ESP_EVENT_ANY_ID, mqtt_event_handler, client);    // 4、開啟MQTT功能    esp_mqtt_client_start(client);}void app_main(void){    ESP_LOGI(TAG, "[APP] Startup..");    ESP_LOGI(TAG, "[APP] Free memory: %d bytes", esp_get_free_heap_size());    ESP_LOGI(TAG, "[APP] IDF version: %s", esp_get_idf_version());    esp_log_level_set("*", ESP_LOG_INFO);    esp_log_level_set("MQTT_CLIENT", ESP_LOG_VERBOSE);    esp_log_level_set("MQTT_EXAMPLE", ESP_LOG_VERBOSE);    esp_log_level_set("TRANSPORT_TCP", ESP_LOG_VERBOSE);    esp_log_level_set("TRANSPORT_SSL", ESP_LOG_VERBOSE);    esp_log_level_set("TRANSPORT", ESP_LOG_VERBOSE);    esp_log_level_set("OUTBOX", ESP_LOG_VERBOSE);    ESP_ERROR_CHECK(nvs_flash_init());    ESP_ERROR_CHECK(esp_netif_init());    ESP_ERROR_CHECK(esp_event_loop_create_default());    /* This helper function configures Wi-Fi or Ethernet, as selected in menuconfig.     * Read "Establishing Wi-Fi or Ethernet Connection" section in     * examples/protocols/README.md for more information about this function.     */    ESP_ERROR_CHECK(example_connect());    mqtt_app_start();}