FreeRTOS 队列：任务之间怎么安全传数据

上一篇你建出了两个任务，各跑各的死循环，互不干扰。爽是爽，但马上撞上一个绕不开的问题：它俩怎么把数据递给对方？

最典型的场景——一个任务负责读传感器（温度、电量、按键状态），读到的值得交给另一个任务去上报、去显示、去存。两个任务，一份数据要从这边流到那边。

你第一反应大概是：搞个全局变量呗，一个任务写，另一个读。打住——这是新手最容易埋的雷。两个任务被调度器随时切来切去，A 任务写到一半（比如刚写完结构体的前半个字段），CPU 被切给 B 任务，B 读到的就是个写了一半的脏数据。这叫竞态（race condition），出的 bug 飘忽、偶发、半夜复现一次再也抓不住，能把你逼疯。

这一节给正解：FreeRTOS 队列（Queue）。它是任务间传数据的首选工具——线程安全、带阻塞、自动排队。你把数据塞进队列一头，另一头的任务取走，整个过程的加锁、同步全由 FreeRTOS 替你扛了，你碰都不用碰那些底层细节。

读这篇前，你需要先跑通过上一篇的 FreeRTOS 任务——会用 xTaskCreate 建任务、懂栈和优先级、知道任务体必须 for(;;) 死循环。本篇不接任何线，传感器数据用代码模拟（一个递增的假读数），调试全靠串口看 ESP_LOGI。

---

第一步：先把生产者-消费者跑通

下面这段是完整可烧录的程序——它建一条队列 + 两个任务：生产者任务每秒模拟读一次传感器，把读数打包成一个结构体 xQueueSend 进队列；消费者任务 xQueueReceive 阻塞地等在队列另一头，一收到就 ESP_LOGI 打出来。直接放进工程的 main/main.c：

#include "freertos/FreeRTOS.h"
#include "freertos/task.h"
#include "freertos/queue.h"          // 队列的头文件，别忘了
#include "esp_log.h"

static const char *TAG = "queue";

// 要传的数据：打包成一个结构体一起传，比传一堆散变量干净
typedef struct {
    int   id;            // 第几次读
    float temperature;   // 模拟的温度读数
} sensor_data_t;

// 队列句柄——两个任务靠它认同一条队列
static QueueHandle_t sensor_queue = NULL;

// 生产者：模拟读传感器，每秒往队列塞一条
void producer_task(void *arg)
{
    int count = 0;
    for (;;) {
        sensor_data_t data = {
            .id = count,
            .temperature = 25.0f + count * 0.5f,   // 假装温度在慢慢升
        };
        // 把 data 这一项发进队列；满了就最多等 100ms
        if (xQueueSend(sensor_queue, &data, pdMS_TO_TICKS(100)) == pdPASS) {
            ESP_LOGI(TAG, "生产 #%d，温度 %.1f，已入队", data.id, data.temperature);
        } else {
            ESP_LOGW(TAG, "队列满了，#%d 没塞进去", data.id);  // 等满 100ms 还没空位
        }
        count++;
        vTaskDelay(pdMS_TO_TICKS(1000));
    }
}

// 消费者：阻塞地等队列来数据，来了就处理
void consumer_task(void *arg)
{
    sensor_data_t received;
    for (;;) {
        // portMAX_DELAY = 队列空就一直睡着等，不耗 CPU，来数据才醒
        if (xQueueReceive(sensor_queue, &received, portMAX_DELAY) == pdPASS) {
            ESP_LOGI(TAG, "  ↳ 消费 #%d，处理温度 %.1f", received.id, received.temperature);
        }
    }
}

void app_main(void)
{
    // 建队列：最多排 5 项，每项是一个 sensor_data_t 那么大
    sensor_queue = xQueueCreate(5, sizeof(sensor_data_t));
    if (sensor_queue == NULL) {
        ESP_LOGE(TAG, "队列建失败，八成是堆内存不够");
        return;
    }

    xTaskCreate(producer_task, "producer", 2048, NULL, 5, NULL);
    xTaskCreate(consumer_task, "consumer", 2048, NULL, 5, NULL);
}

在工程目录下一条命令编译、烧录、看日志：

idf.py build flash monitor

（第一次用要先 idf.py set-target esp32s3 选好芯片型号。按 Ctrl + ] 退出监视。）

你应该看到什么

串口里每秒一对日志，生产一条紧跟着消费一条，数据原原本本地从生产者流到了消费者：

I (1010) queue: 生产 #0，温度 25.0，已入队
I (1010) queue:   ↳ 消费 #0，处理温度 25.0
I (2010) queue: 生产 #1，温度 25.5，已入队
I (2010) queue:   ↳ 消费 #1，处理温度 25.5
I (3010) queue: 生产 #2，温度 26.0，已入队
I (3010) queue:   ↳ 消费 #2，处理温度 26.0

盯住这个画面想一秒：生产者把一个 sensor_data_t 塞进队列，消费者在另一头几乎同时就取到了，id 和 temperature 一个字节都没错、没乱。这中间你没写一行加锁代码、没碰任何全局共享变量——队列把竞态那摊脏活全包圆了。这就是它比"裸传全局变量"强的地方：安全是默认的，不是你小心翼翼换来的。

还有个关键细节：消费者用的是 portMAX_DELAY，队列空的时候它不是在空转死等，而是被调度器挂起睡着了，一点 CPU 都不耗；生产者一塞数据进来，它立刻被唤醒。这正是上一篇说的"该睡就让出 CPU"的肌肉记忆——队列的阻塞等待天然就是让出 CPU 的，不会去触发看门狗。

---

第二步：把这三行 API 讲透

队列就靠三个函数撑起来，拆开看：

① xQueueCreate(length, itemSize) —— 先划好这条流水线

sensor_queue = xQueueCreate(5, sizeof(sensor_data_t));
//                          ①最多排几项  ②每一项多少字节

• 第一个参数 5 是队列长度——这条队列最多能囤 5 个还没被取走的项。满了再塞，就得等（看 timeout）。
• 第二个参数是每一项的字节数，这里直接用 sizeof(sensor_data_t) 让编译器算，别手填数字——结构体加个字段你忘了改，立马错位。

建队列时 FreeRTOS 就按 length × itemSize 一次性把内存（5 × 8 = 40 字节左右）划出来了。所以队列吃多少内存在建的那一刻就定死了，跟你实际塞几项无关。长度别拍脑袋给大，几百项的队列在 ESP32-S3 上很费内存。

返回 NULL 就是没建成（堆不够），上面代码里判了——这个判断别省，建失败你还往里塞，就是往空指针上写，当场崩。

② xQueueSend(queue, &item, timeout) —— 塞一项进去（值拷贝！）

xQueueSend(sensor_queue, &data, pdMS_TO_TICKS(100));
//         哪条队列      数据地址  最多等多久

这里有个全篇最该掰清楚的概念：xQueueSend 传的第二个参数是 &data（地址），但它干的事是值拷贝——FreeRTOS 会把 data 这块内存里的内容整个复制一份进队列，存的是副本，不是 data 这个变量本身。

这意味着什么？意味着 xQueueSend 一返回，你那个 data 局部变量就可以随便改、随便销毁了，队列里存的是它当时的快照，井水不犯河水。上面生产者每轮都用一个新的局部 data，完全没问题——发完它就被回收，但队列里的副本好好的。

③ xQueueReceive(queue, &buf, timeout) —— 取一项出来（也是值拷贝）

xQueueReceive(sensor_queue, &received, portMAX_DELAY);
//            哪条队列       收到哪    最多等多久

对称地，xQueueReceive 把队列头部那一项拷进你给的 &received 这块内存，然后从队列里移除它（队列是先进先出 FIFO，谁先进谁先被取走）。取出来的是副本，所以你拿到 received 后想怎么用怎么用，不影响别人。

timeout：两个极端最常用

xQueueSend 和 xQueueReceive 最后那个参数都是 timeout（最多等多久），单位是 tick，用 pdMS_TO_TICKS(毫秒) 换算。它决定了队列满（发不进）/ 空（取不到）时这个调用怎么办：

• portMAX_DELAY（无限等）：发现满了/空了就一直阻塞睡着，直到有空位/有数据才醒。消费者绝大多数用这个——平时就该躺着等数据，不耗 CPU。
• 0（不等，立即返回）：满了/空了立刻返回失败（pdFALSE），一秒都不等。适合"有就拿、没有就算了、我还有别的活干"的场景。
• pdMS_TO_TICKS(100)（等一会儿）：最多等 100ms，等到了返回成功，等满了还不行就返回失败。上面生产者用的就是这个——队列满了愿意等一下，但不死等。

怎么判断成功？ xQueueSend/xQueueReceive 返回 pdPASS（成功）或 pdFALSE（超时/失败）。这个返回值要判，尤其用了 0 或有限 timeout 时——不判的话，发失败了你以为发出去了，数据悄无声息地丢了。

---

第三步：为什么深一层——队列满了、空了到底发生什么

把队列想成一条有固定座位数的传送带（座位数 = 你建队列时给的 length）。理解满/空的行为，就理解了队列的脾气。

队列空时，xQueueReceive 的行为：传送带上没货，取的人来了——

• timeout 是 portMAX_DELAY：站着等，睡过去，来货立刻醒（最常用，零 CPU 消耗）。
• timeout 是 0：扭头就走，返回 pdFALSE。
• 有限 timeout：等到点，等不到就走。

队列满时，xQueueSend 的行为：传送带座位坐满了，送货的来了——

• timeout 是 portMAX_DELAY：站着等，等到有人取走腾出座位才塞进去。
• timeout 是 0：塞不进，返回 pdFALSE——这一项就丢了（除非你自己留着重发）。
• 有限 timeout：等一会儿，腾出来就塞，等满还没空位就返回失败。

这里藏着一个产品级的设计判断：生产快、消费慢，队列迟早会满，满了你想怎么办？ 三种策略各有用处——

1. 死等（portMAX_DELAY）：生产者被"反压"住，跟着消费者节奏走，一条不丢。适合数据一条都不能少的场景（比如指令）。但代价是生产者会被拖慢。
2. 丢弃（timeout 0 + 判返回值）：满了就丢最新的，生产者不被拖慢。适合"丢几条无所谓、要的是实时最新值"的场景（比如显示用的温度，丢一帧下一帧就刷上了）。
3. 加大队列长度：治标，只是把"什么时候满"往后推。真正的根因往往是消费者太慢，得去优化消费者、或拉高它的优先级。

---

第四步：中断里发数据，必须用 FromISR 版本

到这有个绕不开的实战场景：数据往往是中断里产生的。回想上一篇 L2 的中断篇——按键一按触发 ISR、定时器到点触发 ISR。你很自然会想：那我在 ISR 里直接 xQueueSend 把事件发给某个任务处理，行不行？

不行。在中断服务程序（ISR）里，绝对不能调普通版的 xQueueSend，必须用专门的 xQueueSendFromISR。

为什么？普通版 xQueueSend 里面可能会阻塞（队列满了要等），而中断上下文里绝对不能阻塞——ISR 是抢在所有任务前面跑的，它要是睡过去，整个系统都得跟着卡死。所以 FreeRTOS 给了一套 ...FromISR 的中断专用版：它们从不阻塞（满了就直接返回失败，不等），用法也略有不同：

#include "freertos/FreeRTOS.h"
#include "freertos/queue.h"
#include "driver/gpio.h"

static QueueHandle_t evt_queue = NULL;

// 中断服务程序：按键一按就触发，把引脚号发进队列
static void IRAM_ATTR button_isr(void *arg)
{
    uint32_t gpio = (uint32_t)arg;
    BaseType_t hpw = pdFALSE;                    // 高优先级任务是否被唤醒的标志

    // ISR 专用版：不带阻塞 timeout，多一个 &hpw 出参
    xQueueSendFromISR(evt_queue, &gpio, &hpw);

    // 收尾必做：如果发数据唤醒了更高优先级的任务，立刻切过去
    portYIELD_FROM_ISR(hpw);
}

// 任务侧：照常用普通的 xQueueReceive 等
void button_task(void *arg)
{
    uint32_t gpio;
    for (;;) {
        if (xQueueReceive(evt_queue, &gpio, portMAX_DELAY) == pdPASS) {
            ESP_LOGI(TAG, "GPIO %lu 上有按键，在任务里慢慢处理", gpio);
        }
    }
}

两处和普通版不一样，记牢：

• xQueueSendFromISR(q, &item, &hpw)：没有 timeout 参数（ISR 不能等），最后多一个出参 &hpw。FreeRTOS 通过它告诉你："你这一发，把一个比当前更高优先级的任务唤醒了"。
• ISR 结尾必须 portYIELD_FROM_ISR(hpw)：如果 hpw 被置成了 pdTRUE（确实唤醒了高优先级任务），这行会让中断退出时立刻切到那个被唤醒的任务，而不是傻等回到原来被打断的低优先级任务。漏了这行，被唤醒的任务可能要等到下一个调度时机才跑——延迟就上来了。

这套"ISR 收事件 → 扔进队列 → 任务里慢慢处理"的模式，是嵌入式的经典套路：ISR 里只干最快的活（把事件扔队列就走），重活留给任务。中断越短越好，长逻辑全甩给队列另一头的任务，系统才稳。

---

故障排查：队列不工作，按这个查

现象 / 日志	最可能的原因	怎么办
数据时不时丢几条，xQueueSend 返回 pdFALSE	队列满了：生产比消费快，timeout 给了 0 或太短	看清需求：要不丢就用 portMAX_DELAY 死等；要实时就接受丢；或加大长度/提速消费者
消费者收到的数据是乱码 / 错位（字段对不上）	xQueueCreate 的 itemSize 和实际发的类型不符	建队列、发、收三处必须是同一个类型；itemSize 一律写 sizeof(类型)，别手填数字
在 ISR 里调了 xQueueSend，系统崩溃 / 重启	中断里用了普通版（可能阻塞），违规	ISR 里只能用 xQueueSendFromISR，结尾配 portYIELD_FROM_ISR(hpw)
消费者收到数据了，但内容是发送时之后被改过的值	你传的是指针，且指针指向的数据被改/被回收了	小数据用值拷贝（直接发结构体）最稳；非传指针不可，则保证那块数据活得够久（见变体一的坑）
xQueueCreate 返回 NULL，建不出来	堆内存不够（length × itemSize 太大）	把队列长度削到够用；ESP_LOGI 打 esp_get_free_heap_size() 看剩多少堆
消费者任务一直收不到（卡在 xQueueReceive）	生产者根本没发成功，或发错了队列句柄	在生产者侧判 xQueueSend 返回值；确认两个任务用的是同一个队列句柄变量

---

变体：两个进阶玩法

变体一：大数据传指针，而不是值拷贝

值拷贝有个边界：项太大就别拷了。比如你要传一帧 1KB 的图像缓冲、一个几百字节的大结构体——队列每发一次就整块拷贝一遍，又慢又费内存（队列本身还得按这么大划空间）。

这时候改成传指针：队列里只存一个指针（4 字节），数据本体放在别处，靠指针引用。

// 队列里存的是指针，不是数据本体
QueueHandle_t q = xQueueCreate(5, sizeof(big_buffer_t *));   // 注意是指针的大小

// 发送方：把指针发进去
big_buffer_t *buf = malloc(sizeof(big_buffer_t));
fill_buffer(buf);
xQueueSend(q, &buf, portMAX_DELAY);     // 拷的是这个指针的值（4字节），不是 1KB 数据

// 接收方：拿到指针，用完负责 free
big_buffer_t *got;
xQueueReceive(q, &got, portMAX_DELAY);
process(got);
free(got);                              // 谁用完谁释放，约定好别漏

变体二：多个生产者，一个消费者

队列天生支持多对一：好几个任务往同一条队列发，一个消费者统一收。比如温度任务、电量任务、按键任务都把各自的事件发进同一条队列，一个处理任务挨个取出来分发处理。

关键是让数据自带身份——结构体里放个 type 字段标明来源，消费者按 type 分流：

typedef struct {
    enum { EVT_TEMP, EVT_BATTERY, EVT_BUTTON } type;
    int value;
} event_t;

// 消费者按 type 分流处理
void dispatcher_task(void *arg)
{
    event_t e;
    for (;;) {
        if (xQueueReceive(evt_q, &e, portMAX_DELAY) == pdPASS) {
            switch (e.type) {
                case EVT_TEMP:    ESP_LOGI(TAG, "温度 %d", e.value); break;
                case EVT_BATTERY: ESP_LOGI(TAG, "电量 %d", e.value); break;
                case EVT_BUTTON:  ESP_LOGI(TAG, "按键 %d", e.value); break;
            }
        }
    }
}

队列对多个并发的发送方也是线程安全的——三个任务同时 xQueueSend，FreeRTOS 内部排好序，不会互相踩。这就是事件驱动固件的常见骨架：各路事件汇进一条队列，一个调度任务统一消化。

本篇代码为参考实现，需结合你所用的最新 ESP-IDF / FreeRTOS 文档与实物自校。尤其 xQueueSendFromISR 的参数、portYIELD_FROM_ISR 的用法随版本/配置可能微调，以官方 FreeRTOS 队列文档为准。

---

动手挑战

别只看，动手改一个：

1. 把第一步生产者里的 vTaskDelay 从 1000ms 改成 100ms（生产快 10 倍），同时把消费者里加一行 vTaskDelay(pdMS_TO_TICKS(1000)) 模拟"处理很慢"。观察队列从填满到生产者打出 队列满了 警告的过程——这是你亲手制造的一次"生产快于消费"。然后把生产者的 timeout 从 pdMS_TO_TICKS(100) 改成 portMAX_DELAY，看生产者是不是被"反压"住、变成跟着消费者节奏走、一条不丢。
2. 用变体二的多生产者模式，建两个生产者任务（一个发"温度"事件、一个发"电量"事件，周期不同）+ 一个消费者，往同一条队列发，消费者按 type 分流打日志。验证两路数据混在一条队列里也不会串。

卡住了？把你的代码、idf.py monitor 的完整日志、想要的效果一起发给 AI，把"我看到了什么、期望看到什么、队列长度和 timeout 给的多少"说清楚，它定位会准得多。

---

小结 · 你现在掌握了什么

• 你知道了任务间传数据不能裸用全局变量（会竞态），FreeRTOS 队列是线程安全的首选工具。
• 你能用 xQueueCreate(length, itemSize) 建队列、xQueueSend(q, &item, timeout) 发、xQueueReceive(q, &buf, timeout) 收，跑通一条生产者-消费者流水线。
• 你搞懂了 xQueueSend/xQueueReceive 传的是值拷贝（副本，发完原变量随便动）、itemSize 必须写 sizeof(类型)，以及大数据该改传指针（并自己管好生命周期）。
• 你理解了 timeout 的两个极端——portMAX_DELAY 死等 / 0 立即返回——和队列满/空时各自的行为，以及"死等反压 / 丢弃 / 加大长度"三种满队列策略。
• 你会在 ISR 里用专用的 xQueueSendFromISR(q, &item, &hpw) + 结尾 portYIELD_FROM_ISR(hpw)，知道为什么中断里不能调普通版。

现在任务能安全地传数据了，但还有一类需求队列不直接解决：怎么让一个任务等另一个任务干完某件事再继续？怎么保护一块只能一个任务同时碰的共享资源（比如一条 I2C 总线）？ 这要靠信号量和互斥锁。下一步学FreeRTOS 任务同步：信号量与互斥锁，把任务协作这块补全。

想看 L3 这一级还有哪些课、整条进阶路线长什么样，回L3 关卡总览或完整路线图。