深入解析FreeRTOS队列：从零拷贝到中断处理与内部运作

队列

队列是什么

队列本质是一块连续内存，被划分成等大小的槽，每个槽存一个数据单元。发送方把数据拷贝进槽，接收方从槽里拷贝出来——两边是独立副本，互不干扰。

三个创建参数：

• 队列长度：最多能存几个数据单元
• 数据单元大小：每个槽占几个字节（sizeof(你的数据类型)）
• 队列缓冲区：FreeRTOS 自动按「长度 × 单元大小」分配内存

默认 FIFO（先进先出），也可以用 xQueueSendToFront() 往队头插入，实现 LIFO。

队列 vs 信号量

对比维度	队列	信号量
核心用途	传数据	同步 / 互斥
存储内容	多个数据单元	计数值（0 或 n）
典型场景	中断 → 任务传数据	事件通知、资源保护

相关例子

零拷贝：传递大数据指针

传递大结构体/数组时，直接拷贝数据效率低。可以在队列里只放指针，接收方通过指针访问数据：

#define BUF_SIZE 100
QueueHandle_t xPtrQueue;

void vGenerateTask(void *pv)
{
    // 必须用 static 或动态分配，确保生命周期覆盖接收方处理期间
    static uint8_t buf[BUF_SIZE];
    for (;;)
    {
        for (int i = 0; i < BUF_SIZE; i++)
            buf[i] = i;

        // 先取指针，再发送指针变量的地址
        uint8_t *pBuf = buf;
        xQueueSend(xPtrQueue, &pBuf, portMAX_DELAY);  // 拷贝的是4字节指针值

        vTaskDelay(pdMS_TO_TICKS(2000));
    }
}

void vConsumeTask(void *pv)
{
    uint8_t *p;
    for (;;)
    {
        if (xQueueReceive(xPtrQueue, &p, portMAX_DELAY) == pdPASS)
        {
            uint32_t sum = 0;
            for (int i = 0; i < BUF_SIZE; i++)
                sum += p[i];
            printf("数据总和：%lu\r\n", sum);
        }
    }
}

// 创建：数据单元大小为指针大小
xPtrQueue = xQueueCreate(3, sizeof(uint8_t *));

任务间传递结构体

传感器采集 → 数据处理，经典生产者-消费者模型：

#include "FreeRTOS.h"
#include "task.h"
#include "queue.h"

typedef struct
{
    uint8_t temp;
    uint8_t humi;
} SensorData_t;

QueueHandle_t xSensorQueue;

void vSensorTask(void *pv)
{
    SensorData_t data;
    uint8_t n = 0;
    for (;;)
    {
        data.temp = 25 + n % 10;
        data.humi = 60 + n % 20;
        n++;
        xQueueSend(xSensorQueue, &data, portMAX_DELAY);
        vTaskDelay(pdMS_TO_TICKS(1000));
    }
}

void vProcessTask(void *pv)
{
    SensorData_t d;
    for (;;)
    {
        if (xQueueReceive(xSensorQueue, &d, portMAX_DELAY) == pdPASS)
        {
            printf("温度 %d℃  湿度 %d%%\r\n", d.temp, d.humi);
        }
    }
}

// main 中创建队列和任务
xSensorQueue = xQueueCreate(5, sizeof(SensorData_t));
xTaskCreate(vSensorTask,  "Sensor",  128, NULL, 2, NULL);
xTaskCreate(vProcessTask, "Process", 128, NULL, 3, NULL);
vTaskStartScheduler();

中断 → 任务（串口接收）

QueueHandle_t xUartQueue;

// 中断服务函数：快进快出，只往队列塞数据
void USART1_IRQHandler(void)
{
    BaseType_t xWoken = pdFALSE;
    if (USART_GetITStatus(USART1, USART_IT_RXNE) != RESET)
    {
        uint8_t ch = USART_ReceiveData(USART1);
        xQueueSendFromISR(xUartQueue, &ch, &xWoken);
        USART_ClearITPendingBit(USART1, USART_IT_RXNE);
    }
    portYIELD_FROM_ISR(xWoken);  // 若唤醒了更高优先级任务，立即切换
}

// 处理任务：在任务上下文做真正的业务逻辑
void vUartTask(void *pv)
{
    uint8_t ch;
    for (;;)
    {
        if (xQueueReceive(xUartQueue, &ch, portMAX_DELAY) == pdPASS)
        {
            printf("收到字节：0x%02X\r\n", ch);
        }
    }
}

工作原理

内存结构

FreeRTOS 用读写指针理队列，同时记录已用槽数和空闲槽数，发送/接收时原子更新，保证线程安全。

发送与接收流程

阻塞模式

发送/接收时可指定 xTicksToWait 阻塞等待：

配置	行为	适用场景
0	立即返回，不等	中断中操作队列（必须用这个）
n ticks	最多等 n 个 tick，超时返回失败	有超时容忍的场景
portMAX_DELAY	永久阻塞直到成功	接收任务等待数据

核心 API

创建队列

// 函数原型
QueueHandle_t xQueueCreate(UBaseType_t uxQueueLength, UBaseType_t uxItemSize);

// 存结构体数据的队列
typedef struct {
    uint8_t temp;
    uint8_t humi;
} SensorData_t;

QueueHandle_t xSensorQueue;
xSensorQueue = xQueueCreate(5, sizeof(SensorData_t));
if (xSensorQueue == NULL) {
    // 内存不足，队列创建失败
    configASSERT(0);
}

静态分配版本：资源受限场景可用 xQueueCreateStatic() 避免动态内存碎片：

// 静态创建：内存由调用方提供，不依赖 heap
#define QUEUE_LEN   5
static StaticQueue_t xQueueBuffer;
static uint8_t       ucQueueStorage[ QUEUE_LEN * sizeof(SensorData_t) ];

xSensorQueue = xQueueCreateStatic(QUEUE_LEN, sizeof(SensorData_t),
                                   ucQueueStorage, &xQueueBuffer);
// 静态创建不会返回 NULL，无需判断

发送数据

BaseType_t xQueueSend(QueueHandle_t xQueue, const void *pvItemToQueue, TickType_t xTicksToWait);

参数说明：

参数	含义
xQueue	目标队列句柄
pvItemToQueue	待发送数据的指针，FreeRTOS 会将其内容 memcpy 进队列槽，调用后原变量可随意修改
xTicksToWait	队列满时的最大等待 Tick 数；0 表示不等待立即返回，portMAX_DELAY 表示永久等待

返回值： 数据成功写入队列返回 pdPASS，超时或队列满（xTicksToWait == 0）返回 errQUEUE_FULL。

xQueueSend 等价于 xQueueSendToBack，数据追加到队尾（FIFO）。若需插队到队头，使用 xQueueSendToFront。

底层做了什么？ xQueueSend 实际是宏，展开后调用 xQueueGenericSend()。整个函数的流程如：

对应源码的关键片段如：

for( ; ; )
{
    taskENTER_CRITICAL();
    {
        if( pxQueue->uxMessagesWaiting < pxQueue->uxLength )  /* 有空位 */
        {
            /* 数据拷贝 */
            prvCopyDataToQueue( pxQueue, pvItemToQueue, xCopyPosition );

            /* 唤醒正在等数据的接收任务 */
            if( !listLIST_IS_EMPTY( &pxQueue->xTasksWaitingToReceive ) )
                xTaskRemoveFromEventList( &pxQueue->xTasksWaitingToReceive );

            taskEXIT_CRITICAL();
            return pdPASS;
        }
        else/* 队列满了 */
        {
            if( xTicksToWait == 0 )
            {
                taskEXIT_CRITICAL();
                return errQUEUE_FULL;       /* 告诉调用者：满了 */
            }
            vTaskInternalSetTimeOutState( &xTimeOut );
        }
    }
    taskEXIT_CRITICAL();

    vTaskSuspendAll();
    prvLockQueue( pxQueue );

    if( xTaskCheckForTimeOut( &xTimeOut, &xTicksToWait ) == pdFALSE )
    {
        /* 还没超时，把自己挂到"等待发送"链表，让出 CPU */
        vTaskPlaceOnEventList( &pxQueue->xTasksWaitingToSend, xTicksToWait );
        prvUnlockQueue( pxQueue );
        xTaskResumeAll();
    }
    else
    {
        prvUnlockQueue( pxQueue );
        xTaskResumeAll();
        return errQUEUE_FULL;   /* 超时了，放弃 */
    }
}

xQueueSend 的几种常见用法：

1. 发送基本类型，不阻塞（队列满了就丢弃）

   uint8_t val = 0xA5;
   if (xQueueSend(xSensorQueue, &val, 0) != pdPASS) {
       // 队列满，数据没发出去，按业务需求决定是丢弃还是重试
   }

2. 发送结构体，无限等待（适合生产者不允许丢数据的场景）

   SensorData_t data = { .temp = 26, .humi = 65 };
   xQueueSend(xSensorQueue, &data, portMAX_DELAY);

3. 有限等待 100ms，超时当发送失败处理

   if (xQueueSend(xSensorQueue, &data, pdMS_TO_TICKS(100)) != pdPASS) {
       printf("发送超时\r\n");
   }

若要实现 LIFO（新数据优先被取走），用 xQueueSendToFront：

SensorData_t urgentData = { .temp = 99, .humi = 99 };
xQueueSendToFront(xSensorQueue, &urgentData, 0);

接收数据

BaseType_t xQueueReceive(QueueHandle_t xQueue, void *pvBuffer, TickType_t xTicksToWait);

参数说明：

参数	含义
xQueue	目标队列句柄
pvBuffer	接收缓冲区指针，数据将被拷贝到此处，大小须 ≥ 创建队列时指定的 uxItemSize
xTicksToWait	队列为空时的最大等待 Tick 数；0 表示不等待立即返回，portMAX_DELAY 表示永久等待

返回值： 成功取到数据返回 pdPASS，超时或队列为空（xTicksToWait == 0）返回 errQUEUE_EMPTY。

内部执行流程

结合源码（queue.c: xQueueReceive），整个过程是一个带超时的循环：

对应关键源码：

for( ; ; )
{
    taskENTER_CRITICAL();
    {
        if( pxQueue->uxMessagesWaiting > 0 )  /* 队列有数据 */
        {
            /* 把队头数据 memcpy 到 pvBuffer，pcReadFrom 指针后移 */
            prvCopyDataFromQueue( pxQueue, pvBuffer );
            pxQueue->uxMessagesWaiting--;

            /* 取出数据腾出了空位，唤醒因队满而阻塞的发送任务 */
            if( !listLIST_IS_EMPTY( &pxQueue->xTasksWaitingToSend ) )
                xTaskRemoveFromEventList( &pxQueue->xTasksWaitingToSend );

            taskEXIT_CRITICAL();
            return pdPASS;
        }
        else/* 队列为空 */
        {
            if( xTicksToWait == 0 )
            {
                taskEXIT_CRITICAL();
                return errQUEUE_EMPTY;      /* 告知调用者：空了 */
            }
            vTaskInternalSetTimeOutState( &xTimeOut );  /* 记录开始等待的时间戳（仅首次） */
        }
    }
    taskEXIT_CRITICAL();

    vTaskSuspendAll();
    prvLockQueue( pxQueue );

    if( xTaskCheckForTimeOut( &xTimeOut, &xTicksToWait ) == pdFALSE )
    {
        /* 还没超时，把自己挂到"等待接收"链表，让出 CPU */
        vTaskPlaceOnEventList( &pxQueue->xTasksWaitingToReceive, xTicksToWait );
        prvUnlockQueue( pxQueue );
        xTaskResumeAll();
    }
    else
    {
        prvUnlockQueue( pxQueue );
        xTaskResumeAll();
        return errQUEUE_EMPTY;  /* 超时，放弃等待 */
    }
}

典型用法

1. 阻塞等待，直到取到数据（最常见写法）

   SensorData_t rxData;
   if (xQueueReceive(xSensorQueue, &rxData, portMAX_DELAY) == pdPASS)
   {
       printf("温度：%d℃，湿度：%d%%\r\n", rxData.temp, rxData.humi);
   }

2. 带超时：等 100ms，超时则执行兜底逻辑

   if (xQueueReceive(xSensorQueue, &rxData, pdMS_TO_TICKS(100)) == pdPASS)
   {
       process(&rxData);
   }
   else
   {
       // 100ms 内没收到数据，可能传感器故障
       log_warn("sensor timeout");
   }

3. 非阻塞轮询（xTicksToWait = 0）

   if (xQueueReceive(xSensorQueue, &rxData, 0) == pdPASS)
   {
       // 有数据就处理，没有就跳过
   }

xQueuePeek：只看不取

// xQueuePeek：只看不取，数据仍留在队列中
SensorData_t peekData;
if (xQueuePeek(xSensorQueue, &peekData, 0) == pdPASS)
{
    // 数据依然在队列里，下次 Receive 仍能取到
    printf("队头温度预览：%d℃\r\n", peekData.temp);
}

xQueuePeek 与 xQueueReceive 流程几乎相同，区别在于拷贝完数据后会把 pcReadFrom 恢复到原位，数据不会被消耗。适合"先检查、再决定要不要取"的场景，但多任务下需小心竞争（Peek 之后另一个任务可能先 Receive 走了）。

查询队列状态

// 查询队列当前状态
UBaseType_t waiting  = uxQueueMessagesWaiting(xSensorQueue);  // 已有几个数据
UBaseType_t spaces   = uxQueueSpacesAvailable(xSensorQueue);  // 还剩几个空位
printf("队列状态：已用 %u，空闲 %u\r\n", waiting, spaces);

// [中断](https://yunpan.plus/f/34-1)上下文查询已用数量
UBaseType_t waitingISR = uxQueueMessagesWaitingFromISR(xSensorQueue);

中断安全 API

中断中必须使用带 FromISR 后缀的版本，且 xTicksToWait 固定为 0：

// 函数原型
BaseType_t xQueueSendFromISR(QueueHandle_t xQueue,
                             const void *pvItemToQueue,
                             BaseType_t *pxHigherPriorityTaskWoken);

BaseType_t xQueueReceiveFromISR(QueueHandle_t xQueue,
                                void *pvBuffer,
                                BaseType_t *pxHigherPriorityTaskWoken);

总结

三句话记核心：

1. 队列 = 数据中转站，支持 FIFO/LIFO，多发多收。
2. 中断里操作队列必须用 FromISR API，且不能阻塞。
3. 零拷贝传指针，注意内存生命周期。

队列是FreeRTOS中任务间通信最核心、最灵活的机制之一。理解其内部原理，掌握零拷贝、中断处理等高级用法，对于构建高效可靠的嵌入式系统至关重要。如果你想查看更多关于嵌入式开发或系统设计的深度解析，欢迎访问 云栈社区 交流探讨。