Protocolos de Aplicação IoT 🌐
MQTT - Message Queue Telemetry Transport
O MQTT representa protocolo de mensagens leve projetado especificamente para comunicação máquina-a-máquina em redes com largura de banda limitada ou conexões não confiáveis. Sua arquitetura publish-subscribe desacopla produtores e consumidores de mensagens, permitindo escalabilidade excelente e flexibilidade arquitetural que o tornam escolha ideal para aplicações IoT.
O modelo publish-subscribe do MQTT funciona através de broker central que recebe todas as mensagens publicadas e as encaminha para subscribers interessados. Clients publicam mensagens em tópicos estruturados hierarquicamente, enquanto outros clients subscrevem a estes tópicos para receber mensagens relevantes. Esta indireção elimina necessidade de clients conhecerem uns aos outros, simplificando dramaticamente arquitetura de sistemas distribuídos.
Topics no MQTT usam estrutura hierárquica separada por barras similar a caminhos de arquivo, permitindo organização lógica de mensagens e filtragem sofisticada através de wildcards. Por exemplo, um sistema de monitoramento doméstico pode usar tópicos como casa/sala/temperatura e casa/cozinha/umidade, com subscribers podendo receber todas as mensagens de casa através do wildcard casa/# ou apenas temperaturas com casa/+/temperatura.
📋 Níveis de Qualidade de Serviço MQTT
O MQTT oferece três níveis de Quality of Service que determinam garantias de entrega de mensagens, permitindo que você balance confiabilidade contra overhead de rede conforme requisitos de cada mensagem específica. A escolha apropriada de QoS é decisão arquitetural importante que impacta tanto confiabilidade quanto eficiência do sistema.
QoS 0, conhecido como at most once, oferece semântica de entrega best-effort onde mensagem é enviada uma única vez sem confirmação. Este é o nível mais eficiente em termos de largura de banda e latência, apropriado para dados de telemetria de alta frequência onde perda ocasional de amostras individuais é aceitável. Sensores de temperatura transmitindo leituras a cada segundo tipicamente usam QoS 0 já que próxima leitura virá brevemente.
QoS 1, ou at least once, garante que mensagem será entregue pelo menos uma vez através de sistema de confirmação. O sender retransmite mensagem até receber acknowledgment do broker, depois o broker retransmite para subscribers com confirmações similares. Este nível pode resultar em mensagens duplicadas que aplicação deve estar preparada para lidar, mas garante que nenhuma mensagem crítica se perca durante falhas transitórias de rede.
QoS 2, designado exactly once, implementa handshake de quatro vias que garante entrega única de cada mensagem sem duplicatas. Este é o nível mais confiável mas também mais caro em termos de largura de banda e latência, apropriado apenas para mensagens críticas onde duplicação seria problemática, como comandos de controle que não devem ser executados múltiplas vezes ou transações financeiras.
#include "freertos/FreeRTOS.h"
#include "freertos/task.h"
#include "freertos/event_groups.h"
#include "esp_system.h"
#include "esp_log.h"
#include "mqtt_client.h"
#include <string.h>
static const char *TAG = "MQTT_EXAMPLE";
// Configuração do broker MQTT
#define MQTT_BROKER_URI "mqtt://broker.hivemq.com"
#define MQTT_PORT 1883
#define MQTT_CLIENT_ID "ESP32_Device_001"
// Tópicos MQTT
#define TOPIC_BASE "iot/esp32/device001"
#define TOPIC_TEMPERATURE TOPIC_BASE "/sensors/temperature"
#define TOPIC_HUMIDITY TOPIC_BASE "/sensors/humidity"
#define TOPIC_COMMAND TOPIC_BASE "/commands"
#define TOPIC_STATUS TOPIC_BASE "/status"
// Handle do cliente MQTT
static esp_mqtt_client_handle_t mqtt_client = NULL;
static bool mqtt_connected = false;
/**
* Handler de eventos MQTT
*
* Esta função é chamada automaticamente quando eventos MQTT ocorrem,
* como conexão estabelecida, mensagens recebidas, ou desconexões
*/
static void mqtt_event_handler(void *handler_args, esp_event_base_t base,
int32_t event_id, void *event_data)
{
esp_mqtt_event_handle_t event = event_data;
esp_mqtt_client_handle_t client = event->client;
switch ((esp_mqtt_event_id_t)event_id) {
case MQTT_EVENT_CONNECTED:
ESP_LOGI(TAG, "MQTT conectado ao broker");
mqtt_connected = true;
// Publicar mensagem de status online
esp_mqtt_client_publish(client, TOPIC_STATUS, "online", 0, 1, 1);
// Subscrever tópicos de interesse
int msg_id = esp_mqtt_client_subscribe(client, TOPIC_COMMAND, 1);
ESP_LOGI(TAG, "Subscrito ao tópico %s (msg_id=%d)",
TOPIC_COMMAND, msg_id);
// Subscrever com wildcard para todos os sensores
msg_id = esp_mqtt_client_subscribe(client, TOPIC_BASE "/sensors/#", 0);
ESP_LOGI(TAG, "Subscrito aos sensores (msg_id=%d)", msg_id);
break;
case MQTT_EVENT_DISCONNECTED:
ESP_LOGW(TAG, "MQTT desconectado");
mqtt_connected = false;
break;
case MQTT_EVENT_SUBSCRIBED:
ESP_LOGI(TAG, "Subscrição confirmada (msg_id=%d)", event->msg_id);
break;
case MQTT_EVENT_UNSUBSCRIBED:
ESP_LOGI(TAG, "Cancelamento de subscrição confirmado (msg_id=%d)",
event->msg_id);
break;
case MQTT_EVENT_PUBLISHED:
ESP_LOGD(TAG, "Publicação confirmada (msg_id=%d)", event->msg_id);
break;
case MQTT_EVENT_DATA:
// Mensagem recebida
ESP_LOGI(TAG, "=== Mensagem MQTT Recebida ===");
ESP_LOGI(TAG, "Tópico: %.*s", event->topic_len, event->topic);
ESP_LOGI(TAG, "Dados: %.*s", event->data_len, event->data);
// Processar comando se for do tópico de comandos
if (strncmp(event->topic, TOPIC_COMMAND, event->topic_len) == 0) {
process_mqtt_command(event->data, event->data_len);
}
break;
case MQTT_EVENT_ERROR:
ESP_LOGE(TAG, "Erro MQTT");
if (event->error_handle->error_type == MQTT_ERROR_TYPE_TCP_TRANSPORT) {
ESP_LOGE(TAG, "Erro de transporte TCP");
}
break;
default:
ESP_LOGD(TAG, "Evento MQTT: %d", event->event_id);
break;
}
}
/**
* Inicializar cliente MQTT
*
* Configura todos os parâmetros de conexão e registra handler de eventos
*/
esp_err_t mqtt_app_start(void)
{
ESP_LOGI(TAG, "Inicializando cliente MQTT");
// Configurar cliente MQTT
esp_mqtt_client_config_t mqtt_cfg = {
.broker.address.uri = MQTT_BROKER_URI,
.credentials.client_id = MQTT_CLIENT_ID,
// Last Will and Testament - mensagem enviada se desconexão abrupta
.session.last_will.topic = TOPIC_STATUS,
.session.last_will.msg = "offline",
.session.last_will.msg_len = 7,
.session.last_will.qos = 1,
.session.last_will.retain = 1,
// Keep alive
.network.timeout_ms = 5000,
.session.keepalive = 120,
// Buffers
.buffer.size = 1024,
.buffer.out_size = 1024,
};
// Criar cliente MQTT
mqtt_client = esp_mqtt_client_init(&mqtt_cfg);
if (mqtt_client == NULL) {
ESP_LOGE(TAG, "Falha ao criar cliente MQTT");
return ESP_FAIL;
}
// Registrar handler de eventos
esp_err_t ret = esp_mqtt_client_register_event(mqtt_client,
ESP_EVENT_ANY_ID,
mqtt_event_handler,
NULL);
if (ret != ESP_OK) {
ESP_LOGE(TAG, "Falha ao registrar handler de eventos: %s",
esp_err_to_name(ret));
return ret;
}
// Iniciar cliente
ret = esp_mqtt_client_start(mqtt_client);
if (ret == ESP_OK) {
ESP_LOGI(TAG, "Cliente MQTT iniciado");
} else {
ESP_LOGE(TAG, "Falha ao iniciar cliente MQTT: %s",
esp_err_to_name(ret));
}
return ret;
}
/**
* Publicar leitura de sensor via MQTT
*
* Formata dados e publica no tópico apropriado com QoS especificado
*/
int mqtt_publish_sensor_data(const char *sensor_type, float value, int qos)
{
if (!mqtt_connected) {
ESP_LOGW(TAG, "MQTT desconectado, não é possível publicar");
return -1;
}
char topic[128];
char payload[64];
// Construir tópico e payload
snprintf(topic, sizeof(topic), "%s/sensors/%s", TOPIC_BASE, sensor_type);
snprintf(payload, sizeof(payload), "{\"value\":%.2f,\"unit\":\"%s\"}",
value, sensor_type);
// Publicar mensagem
int msg_id = esp_mqtt_client_publish(mqtt_client, topic, payload, 0, qos, 0);
if (msg_id >= 0) {
ESP_LOGI(TAG, "Publicado: %s = %.2f (msg_id=%d, QoS=%d)",
sensor_type, value, msg_id, qos);
} else {
ESP_LOGE(TAG, "Falha ao publicar dados do sensor");
}
return msg_id;
}
/**
* Publicar dados JSON estruturados
*
* Demonstra envio de mensagens complexas com múltiplos campos
*/
int mqtt_publish_json(const char *topic, const char *json_str, int qos)
{
if (!mqtt_connected) {
return -1;
}
int msg_id = esp_mqtt_client_publish(mqtt_client, topic,
json_str, 0, qos, 0);
if (msg_id >= 0) {
ESP_LOGD(TAG, "JSON publicado no tópico %s (msg_id=%d)", topic, msg_id);
}
return msg_id;
}
/**
* Processar comandos recebidos via MQTT
*
* Implementa lógica para responder a comandos remotos
*/
void process_mqtt_command(const char *command, int length)
{
char cmd_buffer[128];
// Copiar comando para buffer com terminação null
int len = length < sizeof(cmd_buffer) - 1 ? length : sizeof(cmd_buffer) - 1;
memcpy(cmd_buffer, command, len);
cmd_buffer[len] = '\0';
ESP_LOGI(TAG, "Processando comando: %s", cmd_buffer);
// Implementar processamento de comandos específicos
if (strcmp(cmd_buffer, "status") == 0) {
// Enviar status detalhado
char status_json[256];
snprintf(status_json, sizeof(status_json),
"{\"heap_free\":%d,\"uptime\":%lu,\"wifi_rssi\":%d}",
esp_get_free_heap_size(),
(unsigned long)(xTaskGetTickCount() * portTICK_PERIOD_MS / 1000),
-50); // Placeholder para RSSI real
mqtt_publish_json(TOPIC_STATUS, status_json, 1);
} else if (strcmp(cmd_buffer, "reset") == 0) {
ESP_LOGW(TAG, "Comando de reset recebido, reiniciando...");
vTaskDelay(pdMS_TO_TICKS(1000));
esp_restart();
} else {
ESP_LOGW(TAG, "Comando desconhecido: %s", cmd_buffer);
}
}
/**
* Task de publicação periódica de telemetria
*
* Simula coleta e envio regular de dados de sensores
*/
void mqtt_telemetry_task(void *pvParameters)
{
ESP_LOGI(TAG, "Task de telemetria MQTT iniciada");
// Aguardar conexão MQTT
while (!mqtt_connected) {
ESP_LOGI(TAG, "Aguardando conexão MQTT...");
vTaskDelay(pdMS_TO_TICKS(1000));
}
int sample_count = 0;
while (1) {
// Simular leituras de sensores
float temperature = 20.0f + (esp_random() % 100) / 10.0f;
float humidity = 50.0f + (esp_random() % 300) / 10.0f;
// Publicar dados individuais
// Temperatura com QoS 1 (mais importante)
mqtt_publish_sensor_data("temperature", temperature, 1);
vTaskDelay(pdMS_TO_TICKS(100));
// Umidade com QoS 0 (menos crítico)
mqtt_publish_sensor_data("humidity", humidity, 0);
// A cada 10 amostras, publicar estatísticas agregadas
sample_count++;
if (sample_count % 10 == 0) {
char stats_json[256];
snprintf(stats_json, sizeof(stats_json),
"{\"samples\":%d,\"avg_temp\":%.1f,\"avg_hum\":%.1f}",
sample_count, temperature, humidity);
mqtt_publish_json(TOPIC_BASE "/statistics", stats_json, 1);
}
vTaskDelay(pdMS_TO_TICKS(5000)); // Publicar a cada 5 segundos
}
}
/**
* Parar cliente MQTT graciosamente
*
* Publica mensagem de despedida e desconecta limpamente
*/
void mqtt_app_stop(void)
{
if (mqtt_connected) {
ESP_LOGI(TAG, "Publicando mensagem de despedida");
esp_mqtt_client_publish(mqtt_client, TOPIC_STATUS, "offline", 0, 1, 1);
vTaskDelay(pdMS_TO_TICKS(100));
}
esp_mqtt_client_stop(mqtt_client);
esp_mqtt_client_destroy(mqtt_client);
mqtt_client = NULL;
mqtt_connected = false;
ESP_LOGI(TAG, "Cliente MQTT parado");
}
void app_main(void)
{
ESP_LOGI(TAG, "Demonstrando protocolo MQTT");
// Nota: WiFi deve estar conectado antes de usar MQTT
// Veja seção anterior sobre configuração WiFi
// Iniciar cliente MQTT
esp_err_t ret = mqtt_app_start();
if (ret == ESP_OK) {
// Criar task de telemetria
xTaskCreate(mqtt_telemetry_task, "MQTT_Telemetry",
4096, NULL, 5, NULL);
ESP_LOGI(TAG, "Sistema MQTT inicializado");
} else {
ESP_LOGE(TAG, "Falha ao inicializar MQTT");
}
}