基于FreeRTOS的多传感器数据采集系统设计与实现 — 从任务调度到MQTT上云

┌─────────────────────────────────────────────────────────┐
│                    Application Layer                     │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐  │
│  │ CLI 调试  │ │ MQTT 发布 │ │  OTA     │ │ 告警策略 │  │
│  └──────────┘ └──────────┘ └──────────┘ └──────────┘  │
├─────────────────────────────────────────────────────────┤
│                  FreeRTOS Kernel                        │
│  ┌─────┐ ┌──────┐ ┌───────┐ ┌──────┐ ┌───────────┐  │
│  │任务  │ │队列  │ │信号量 │ │事件组│ │软件定时器 │  │
│  │调度器│ │(Queue)│ │(Sem)  │ │(EventGrp)│ │(Timer)   │  │
│  └─────┘ └──────┘ └───────┘ └──────┘ └───────────┘  │
├─────────────────────────────────────────────────────────┤
│              BSP & Driver Layer (HAL)                   │
│  I2C  │  SPI  │  UART  │  GPIO  │  DMA  │  RTC  │  WD │
└─────────────────────────────────────────────────────────┘
│  Hardware: STM32F407 │ SHT30 │ BMP280 │ MPU6050 │ ESP8266 │
└─────────────────────────────────────────────────────────┘

// 传感器原始数据（含时间戳）
typedef struct {
    TickType_t timestamp;
    float temperature;   // SHT30
    float humidity;
    float pressure;      // BMP280
    float accel_x, accel_y, accel_z; // MPU6050
    uint8_t sensor_id;
} SensorRaw_t;

// 队列句柄
QueueHandle_t xSensorDataQueue;
QueueHandle_t xPublishQueue;

// 在 main() 中创建
xSensorDataQueue = xQueueCreate(10, sizeof(SensorRaw_t));
xPublishQueue   = xQueueCreate(5,  sizeof(char *));

void vTaskSensorReader(void *pvParameters) {
    TickType_t xLastWakeTime = xTaskGetTickCount();
    const TickType_t xPeriod = pdMS_TO_TICKS(200);
    SensorRaw_t raw;

    while (1) {
        vTaskDelayUntil(&xLastWakeTime, xPeriod);

        // 读取 SHT30（I2C）
        if (HAL_I2C_Master_Transmit(&hi2c1, SHT30_ADDR, cmd, 2, 100) == HAL_OK) {
            HAL_I2C_Master_Receive(&hi2c1, SHT30_ADDR, rx_buf, 6, 100);
            raw.temperature = sht30_calc_temp(rx_buf);
            raw.humidity    = sht30_calc_humi(rx_buf);
        }

        // 读取 BMP280（SPI）
        bmp280_read(&raw.pressure, &raw.temperature_secondary);

        // 读取 MPU6050（通过 DMA 加速，此处略）
        mpu6050_read_gyro_accel(&raw.accel_x, ...);

        raw.timestamp = xTaskGetTickCount();

        // 发送到处理队列（若队列满则丢弃，采用非阻塞）
        if (xQueueSend(xSensorDataQueue, &raw, 0) != pdPASS) {
            // 可增加丢弃计数统计
        }
    }
}

void vTaskDataProcessor(void *pvParameters) {
    SensorRaw_t raw, filtered;
    static float temp_buf[5] = {0};
    static uint8_t idx = 0;

    while (1) {
        if (xQueueReceive(xSensorDataQueue, &raw, portMAX_DELAY) == pdPASS) {
            // 滑动平均滤波（温度）
            temp_buf[idx++ % 5] = raw.temperature;
            float sum = 0;
            for (int i = 0; i < 5; i++) sum += temp_buf[i];
            filtered.temperature = sum / 5.0f;

            // 异常检测：若温度突变 > 5°C，置标志
            static float last_temp = 0;
            if (fabs(filtered.temperature - last_temp) > 5.0f) {
                filtered.temperature = last_temp; // 剔除野值
                // 可发送事件到告警任务
            }
            last_temp = filtered.temperature;

            // 打包为 JSON 字符串（使用 cJSON 库）
            cJSON *root = cJSON_CreateObject();
            cJSON_AddNumberToObject(root, "temp", filtered.temperature);
            cJSON_AddNumberToObject(root, "humi", raw.humidity);
            cJSON_AddNumberToObject(root, "press", raw.pressure);
            cJSON_AddNumberToObject(root, "ts", raw.timestamp);
            char *json_str = cJSON_PrintUnformatted(root);
            cJSON_Delete(root);

            // 发送到发布队列（若队列满则等待 100ms）
            if (xQueueSend(xPublishQueue, &json_str, pdMS_TO_TICKS(100)) != pdPASS) {
                free(json_str); // 防止内存泄漏
            }
        }
    }
}

#define ESP_CMD_TIMEOUT  pdMS_TO_TICKS(3000)

void vTaskMQTTPublisher(void *pvParameters) {
    char *json_str;
    while (1) {
        if (xQueueReceive(xPublishQueue, &json_str, portMAX_DELAY) == pdPASS) {
            // 构建 MQTT PUBLISH 报文（此处简化，实际需生成固定头部）
            char mqtt_pub[512];
            snprintf(mqtt_pub, sizeof(mqtt_pub),
                "AT+MQTTPUB=0,\"data/%.3f\",\"%s\",0,0\r\n",
                (double)xTaskGetTickCount() / 1000.0, json_str);

            // 通过 DMA 发送至 ESP8266
            HAL_UART_Transmit_DMA(&huart2, (uint8_t*)mqtt_pub, strlen(mqtt_pub));

            // 等待响应（事件组）
            EventBits_t bits = xEventGroupWaitBits(
                xESP8266EventGroup,
                ESP_RESPONSE_BIT | ESP_ERROR_BIT,
                pdTRUE,   // 清除位
                pdFALSE,  // 任一满足
                ESP_CMD_TIMEOUT
            );

            if (bits & ESP_ERROR_BIT) {
                // 重试或进入错误处理
                xEventGroupSetBits(xSystemEventGroup, SYS_ERR_MQTT_FAIL);
            } else if (!(bits & ESP_RESPONSE_BIT)) {
                // 超时，复位 ESP8266
                HAL_GPIO_WritePin(ESP_RST_PIN, GPIO_PIN_RESET);
                vTaskDelay(pdMS_TO_TICKS(100));
                HAL_GPIO_WritePin(ESP_RST_PIN, GPIO_PIN_SET);
            }

            free(json_str); // 释放动态内存
        }
    }
}

TimerHandle_t xWDTimer;

void vWDTCallback(TimerHandle_t xTimer) {
    HAL_IWDG_Refresh(&hiwdg);
    // 同时检查系统事件组是否有致命错误
    if (xEventGroupGetBits(xSystemEventGroup) & SYS_FATAL_MASK) {
        // 触发软复位
        NVIC_SystemReset();
    }
}

// 在 main 中创建
xWDTimer = xTimerCreate("WDT", pdMS_TO_TICKS(500), pdTRUE, NULL, vWDTCallback);
xTimerStart(xWDTimer, 0);

// 按键中断（EXTI）
void HAL_GPIO_EXTI_Callback(uint16_t GPIO_Pin) {
    if (GPIO_Pin == KEY_PIN) {
        BaseType_t xHigherPriorityTaskWoken = pdFALSE;
        xSemaphoreGiveFromISR(xKeySemaphore, &xHigherPriorityTaskWoken);
        portYIELD_FROM_ISR(xHigherPriorityTaskWoken);
    }
}

// 按键处理任务（高优先级）
void vTaskKeyHandler(void *pvParameters) {
    while (1) {
        if (xSemaphoreTake(xKeySemaphore, portMAX_DELAY) == pdPASS) {
            // 防抖：延迟 50ms 后再读
            vTaskDelay(pdMS_TO_TICKS(50));
            if (HAL_GPIO_ReadPin(KEY_PIN) == GPIO_PIN_RESET) {
                // 触发一次即时发布（向发布队列插入当前最新数据）
                // 这里可借助队列集或直接读取全局共享数据（需用互斥量保护）
            }
        }
    }
}

void vApplicationStackOverflowHook(TaskHandle_t xTask, char *pcTaskName) {
    // 写入内部 Flash 日志区，然后复位
    write_log_to_flash("Stack overflow in %s", pcTaskName);
    NVIC_SystemReset();
}

// 连接参数示例
char client_id[128];
char username[128];
char password[128];

sprintf(client_id, "%s%s", PRODUCT_ID, DEVICE_NAME);
// 使用腾讯云官方 HMAC 算法生成 password（此处省略，可参考 qcloud-iot-sdk）

// 通过 AT 指令连接
AT+MQTTCONN=<client_id>,<username>,<password>,<keepalive>