8.1 温湿度采集与 MQTT 上云

📚 本节导读

学习时长: 约 90 分钟
难度级别: ⭐⭐⭐⭐☆
前置知识: 第 4.2 任务管理、第 5.10 Sensors 设备、第 7.2.3 MQTT 协议

🎯 学习目标

• 掌握多任务协作的嵌入式系统设计
• 学会使用 Sensors 框架读取温湿度传感器
• 掌握 MQTT 客户端的使用和数据上报
• 理解任务间数据传递的完整流程

一、项目需求

设计一个温湿度采集终端，定时读取 AHT10 传感器数据，通过 MQTT 协议上报到 OneNET 云平台，同时本地 LED 指示工作状态。

功能需求：

• 每 5 秒采集一次温湿度
• 通过 MQTT 上报到 OneNET 平台
• 本地 LED 闪烁指示运行状态
• Shell 命令行可查询实时数据

二、系统架构

┌─────────────────────────────────────────────────────┐
│                    应用层                             │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐          │
│  │ 传感器任务 │  │ MQTT 任务 │  │ LED 任务  │          │
│  │ (采集)    │  │ (上报)    │  │ (指示)    │          │
│  └─────┬────┘  └────┬─────┘  └────┬─────┘          │
│        │             │             │                 │
│  ┌─────▼────┐  ┌─────▼─────┐  ┌───▼────┐           │
│  │ 消息队列  │  │ MQTT 客户端│  │  GPIO  │           │
│  │ (传递数据)│  │ (网络通信) │  │ (LED)  │           │
│  └──────────┘  └───────────┘  └────────┘           │
├─────────────────────────────────────────────────────┤
│              OneOS 内核 + 设备驱动                    │
│  os_task / os_mq / os_sem / sensor / net / mqtt    │
└─────────────────────────────────────────────────────┘

任务设计：

任务	优先级	栈大小	说明
sensor_task	5	1024	读取 AHT10，发送到消息队列
mqtt_task	4	2048	从消息队列取数据，MQTT 上报
led_task	8	512	LED 闪烁，指示运行状态

三、核心代码实现

3.1 全局定义与消息队列

#include <oneos_config.h>
#include <os_task.h>
#include <os_mq.h>
#include <os_sem.h>
#include <device.h>
#include <sensor.h>
#include <mqtt_client.h>

/* 传感器数据结构 */
typedef struct
{
    float temperature;
    float humidity;
    uint32_t timestamp;
} sensor_data_t;

/* 全局对象 */
static os_mq_t      *g_data_mq;
static os_sem_t     *g_net_ready_sem;
static os_device_t  *g_sensor_dev;
static os_mq_client_t *g_mqtt_client;

3.2 传感器采集任务

void sensor_task(void *parameter)
{
    sensor_data_t data;
    struct os_sensor_data sensor_data;

    /* 等待网络就绪 */
    os_sem_wait(g_net_ready_sem, OS_WAIT_FOREVER);

    while (1)
    {
        /* 读取温湿度传感器 */
        if (os_device_read(g_sensor_dev, 0, &sensor_data, sizeof(sensor_data)) > 0)
        {
            data.temperature = sensor_data.data.temp;
            data.humidity    = sensor_data.data.humi;
            data.timestamp   = os_tick_get();

            os_kprintf("[Sensor] Temp: %.1f C, Humi: %.1f %%\r\n",
                       data.temperature, data.humidity);

            /* 发送到消息队列（非阻塞） */
            os_mq_send(g_data_mq, &data, sizeof(data), OS_NO_WAIT);
        }
        else
        {
            os_kprintf("[Sensor] Read failed!\r\n");
        }

        os_task_msleep(5000);  /* 5 秒采集一次 */
    }
}

3.3 MQTT 上报任务

void mqtt_task(void *parameter)
{
    sensor_data_t data;
    char payload[128];
    os_err_t ret;

    /* 初始化 MQTT 连接 */
    ret = os_mqtt_init(g_mqtt_client);
    ret = os_mqtt_connect(g_mqtt_client);

    if (ret != OS_SUCCESS)
    {
        os_kprintf("[MQTT] Connect failed, retry...\r\n");
        os_task_msleep(3000);
        ret = os_mqtt_connect(g_mqtt_client);
    }

    os_kprintf("[MQTT] Connected to broker!\r\n");

    /* 通知传感器任务网络已就绪 */
    os_sem_post(g_net_ready_sem);

    while (1)
    {
        /* 等待传感器数据 */
        if (os_mq_recv(g_data_mq, &data, sizeof(data), OS_WAIT_FOREVER) == sizeof(data))
        {
            /* 构建 JSON 负载 */
            os_snprintf(payload, sizeof(payload),
                        "{\"temperature\":%.1f,\"humidity\":%.1f,\"timestamp\":%d}",
                        data.temperature, data.humidity, data.timestamp);

            /* 发布到 MQTT Topic */
            ret = os_mqtt_publish(g_mqtt_client, "device/sensor/data",
                                  (uint8_t *)payload, os_strlen(payload), 1);
            if (ret == OS_SUCCESS)
            {
                os_kprintf("[MQTT] Published: %s\r\n", payload);
            }
            else
            {
                os_kprintf("[MQTT] Publish failed: %d\r\n", ret);
            }
        }
    }
}

3.4 LED 指示任务

#define LED_PIN  PA0

void led_task(void *parameter)
{
    os_device_t *led_dev;

    led_dev = os_device_open_s("gpio_led");
    if (led_dev == OS_NULL)
    {
        os_kprintf("[LED] Device not found!\r\n");
        return;
    }

    while (1)
    {
        os_device_control(led_dev, OS_GPIO_CMD_SET_PIN, (void *)LED_PIN);
        os_task_msleep(500);

        os_device_control(led_dev, OS_GPIO_CMD_RESET_PIN, (void *)LED_PIN);
        os_task_msleep(500);
    }
}

3.5 Shell 命令：查询实时数据

#include <shell.h>

static void shell_sensor_status(int argc, char *argv[])
{
    sensor_data_t data;

    if (os_mq_recv(g_data_mq, &data, sizeof(data), OS_NO_WAIT) == sizeof(data))
    {
        os_kprintf("Temperature: %.1f C\r\n", data.temperature);
        os_kprintf("Humidity:    %.1f %%\r\n", data.humidity);
        os_kprintf("Timestamp:   %d\r\n", data.timestamp);
    }
    else
    {
        os_kprintf("No data available\r\n");
    }
}
SH_CMD_EXPORT(status, shell_sensor_status, "Show sensor status");

3.6 主函数：系统初始化

int main(void)
{
    /* 初始化内核对象 */
    g_data_mq = os_mq_create("sensor_mq", sizeof(sensor_data_t), 10);
    g_net_ready_sem = os_sem_create("net_sem", 0, 1);

    /* 打开传感器设备 */
    g_sensor_dev = os_device_open_s("aht10_sensor");
    if (g_sensor_dev == OS_NULL)
    {
        os_kprintf("Sensor device not found!\r\n");
    }

    /* 创建任务 */
    os_task_t *task;

    os_task_init(&task, "sensor", sensor_task, NULL, 1024, 5, 10);
    os_task_startup(task);

    os_task_init(&task, "mqtt", mqtt_task, NULL, 2048, 4, 10);
    os_task_startup(task);

    os_task_init(&task, "led", led_task, NULL, 512, 8, 10);
    os_task_startup(task);

    os_kprintf("=== IoT Sensor Node Started ===\r\n");
    return 0;
}

四、数据流图

  ┌─────────┐    os_device_read()    ┌──────────┐
  │  AHT10  │ ───────────────────→  │  sensor  │
  │ 传感器   │                       │  _task   │
  └─────────┘                       └────┬─────┘
                                         │ os_mq_send()
                                    ┌────▼─────┐
                                    │  消息队列  │
                                    │ sensor_mq │
                                    └────┬─────┘
                                         │ os_mq_recv()
                                    ┌────▼─────┐
                                    │  mqtt    │
                                    │  _task   │
                                    └────┬─────┘
                                         │ os_mqtt_publish()
                                    ┌────▼─────┐
                                    │ OneNET   │
                                    │ 云平台    │
                                    └──────────┘

五、运行效果

# 开发板串口输出
=== IoT Sensor Node Started ===
[MQTT] Connected to broker!
[Sensor] Temp: 25.3 C, Humi: 65.1 %
[MQTT] Published: {"temperature":25.3,"humidity":65.1,"timestamp":12345}
[Sensor] Temp: 25.4 C, Humi: 64.8 %
[MQTT] Published: {"temperature":25.4,"humidity":64.8,"timestamp":17345}

# Shell 命令查询
> status
Temperature: 25.4 C
Humidity:    64.8 %
Timestamp:   17345

📝 本节小结

1. 本项目综合使用了任务管理（3 个任务）、消息队列（任务间通信）、信号量（同步）、传感器设备、MQTT 组件
2. 消息队列实现了传感器任务和 MQTT 任务的解耦
3. 信号量确保 MQTT 连接就绪后才开始采集
4. Shell 命令提供了实时查询能力
5. LED 任务提供直观的运行状态指示

🔗 相关链接

• 4.2 任务管理
• 5.10 Sensors设备
• 7.2.3 MQTT协议
• 第8章 README