嵌入式物联网实战：基于 FreeRTOS 与 ESP32 的实时传感数据采集与 MQTT 上云

#include "freertos/FreeRTOS.h"
#include "freertos/task.h"
#include "freertos/queue.h"
#include "freertos/semphr.h"
#include "dht22.h"   // 假设已有驱动

#define DHT_PIN  GPIO_NUM_4
#define QUEUE_LEN 10

typedef struct {
    float temperature;
    float humidity;
    uint64_t timestamp;  // 微秒级时间戳
} sensor_data_t;

QueueHandle_t sensor_queue;

void sensor_task(void *pvParameters) {
    dht22_t sensor;
    dht22_init(&sensor, DHT_PIN);
    TickType_t last_wake = xTaskGetTickCount();
    const TickType_t period = pdMS_TO_TICKS(2000);

    while (1) {
        sensor_data_t data;
        if (dht22_read(&sensor, &data.temperature, &data.humidity) == ESP_OK) {
            data.timestamp = esp_timer_get_time();
            // 向队列发送，若队列满则等待 100ms（避免阻塞）
            if (xQueueSend(sensor_queue, &data, pdMS_TO_TICKS(100)) != pdPASS) {
                ESP_LOGW("SENSOR", "Queue full, dropping data");
            }
        } else {
            ESP_LOGE("SENSOR", "Read failed");
        }
        vTaskDelayUntil(&last_wake, period);
    }
}

#include "cJSON.h"

void processing_task(void *pvParameters) {
    sensor_data_t raw_buf[3] = {0};
    int idx = 0;
    char json_buf[256];

    while (1) {
        sensor_data_t new_data;
        if (xQueueReceive(sensor_queue, &new_data, portMAX_DELAY) == pdPASS) {
            raw_buf[idx % 3] = new_data;
            idx++;
            if (idx < 3) continue;  // 积累足够样本

            // 计算平均
            float avg_temp = 0, avg_humi = 0;
            for (int i = 0; i < 3; i++) {
                avg_temp += raw_buf[i].temperature;
                avg_humi += raw_buf[i].humidity;
            }
            avg_temp /= 3;
            avg_humi /= 3;

            // 构建 JSON
            cJSON *root = cJSON_CreateObject();
            cJSON_AddNumberToObject(root, "temp", avg_temp);
            cJSON_AddNumberToObject(root, "humi", avg_humi);
            cJSON_AddNumberToObject(root, "ts", new_data.timestamp);
            char *json_str = cJSON_PrintUnformatted(root);
            strncpy(json_buf, json_str, sizeof(json_buf) - 1);
            free(json_str);
            cJSON_Delete(root);

            // 发送到 MQTT 任务（通过另一个队列或直接全局变量，但需加互斥锁）
            // 此处采用全局缓冲区 + 互斥锁（更简单）
            // 实际可使用 xQueueOverwrite 等
        }
    }
}

#include "mqtt_client.h"

esp_mqtt_client_handle_t mqtt_client;
SemaphoreHandle_t data_ready_mutex;
char shared_json[256];

void mqtt_task(void *pvParameters) {
    esp_mqtt_client_config_t mqtt_cfg = {
        .broker.address.uri = "mqtt://your_broker_ip:1883",
        .credentials.username = "device001",
        .credentials.authentication.password = "pass123"
    };
    mqtt_client = esp_mqtt_client_init(&mqtt_cfg);
    esp_mqtt_client_start(mqtt_client);

    TickType_t last_pub = xTaskGetTickCount();
    while (1) {
        // 尝试获取数据锁并复制 JSON
        if (xSemaphoreTake(data_ready_mutex, pdMS_TO_TICKS(100)) == pdTRUE) {
            char local_json[256];
            strcpy(local_json, shared_json);
            xSemaphoreGive(data_ready_mutex);
            // 发布
            int msg_id = esp_mqtt_client_publish(mqtt_client, "/dev/sensor", local_json, 0, 0, 0);
            if (msg_id == -1) {
                ESP_LOGE("MQTT", "Publish failed, will retry");
            }
        }
        vTaskDelayUntil(&last_pub, pdMS_TO_TICKS(5000));
    }
}

#include "esp_task_wdt.h"
// 在初始化时启用，超时 10 秒
esp_task_wdt_init(10, true);
// 将每个任务加入看门狗
esp_task_wdt_add(NULL);  // 当前任务

void app_main(void) {
    sensor_queue = xQueueCreate(QUEUE_LEN, sizeof(sensor_data_t));
    data_ready_mutex = xSemaphoreCreateMutex();
    xTaskCreatePinnedToCore(sensor_task, "sensor", 4096, NULL, 3, NULL, 1);
    xTaskCreatePinnedToCore(processing_task, "process", 8192, NULL, 2, NULL, 0);
    xTaskCreatePinnedToCore(mqtt_task, "mqtt", 6144, NULL, 1, NULL, 1);
    // LED 任务略
    ESP_LOGI("MAIN", "All tasks started");
}