FreeRTOS 队列机制深度解析

在掌握了任务管理和内核调度后，你已经对 FreeRTOS 核心有了深入理解。但在学习过程中，很多人都会有一个困惑：队列究竟是如何实现任务阻塞的？ 队列内容何时到达是不可预知的，任务阻塞后又是如何被精准恢复的？

核心原理速览

其实原理非常直观：队列内部维护了两个链表 xTasksWaitingToSend（发送阻塞链表）和 xTasksWaitingToReceive（接收阻塞链表）。

以“获取元素”为例：

  1. 阻塞阶段：如果队列为空，当前任务 TCB 中的“状态列表项”会加入全局的阻塞链表（延时列表），而其“事件列表项”则会加入到该队列维护的接收阻塞链表中。
  2. 恢复阶段：一旦有新元素入队，队列会从其接收阻塞链表中找到优先级最高的任务，将其从队列阻塞链表中移除，并重新加入就绪链表。

了解了队列的这种同步机制后，后续学习信号量和互斥锁（它们本质上是特殊的队列）将会毫无难度。

1. 队列结构体：xQUEUE

队列的相关源码位于 queue.c 和 queue.h。与 TCB 仅保存栈指针类似，队列结构体也不直接包含存储空间，而是在创建时动态分配。

typedef struct QueueDefinition
{
    int8_t * pcHead;     // 指向队列存储区的起始地址
    int8_t * pcWriteTo;  // 指向下一个写入位置

    union
    {
        QueuePointers_t xQueue;     // 队列特有数据（读指针等）
        SemaphoreData_t xSemaphore; // 信号量特有数据
    } u;

    List_t xTasksWaitingToSend;    // 因队列满而阻塞的发送任务链表
    List_t xTasksWaitingToReceive; // 因队列空而阻塞的接收任务链表

    volatile UBaseType_t uxMessagesWaiting; // 当前队列中的消息数量
    UBaseType_t uxLength;    // 队列最大容量
    UBaseType_t uxItemSize;  // 单个元素的大小

    /* 队列锁：记录调度器挂起期间 这个其实不用太在意 还挺复杂的 */
    volatile int8_t cRxLock;
    volatile int8_t cTxLock; 

    uint8_t ucStaticallyAllocated; // 静态分配标志
} xQUEUE;
typedef xQUEUE Queue_t;

typedef struct QueuePointers
{
    int8_t * pcTail;     // 指向存储区的结束地址
    int8_t * pcReadFrom; // 指向上一个读出的位置
} QueuePointers_t;

2. 队列的创建

我们重点分析动态创建过程。FreeRTOS 通过宏定义简化了调用，但底层统一使用 xQueueGenericCreate。

#define xQueueCreate( uxQueueLength, uxItemSize ) \
    xQueueGenericCreate( ( uxQueueLength ), ( uxItemSize ), ( queueQUEUE_TYPE_BASE ) )

2.1 通用创建函数 xQueueGenericCreate

该函数通过 ucQueueType 参数区分基本队列、信号量或互斥锁。

QueueHandle_t xQueueGenericCreate( const UBaseType_t uxQueueLength, const UBaseType_t uxItemSize, const uint8_t ucQueueType )
{
    Queue_t * pxNewQueue = NULL;
    size_t xQueueSizeInBytes;
    uint8_t * pucQueueStorage;

    // 严谨的边界检查：确保内存申请不会溢出
    if( ( uxQueueLength > ( UBaseType_t ) 0 ) && 
        ( ( SIZE_MAX / uxQueueLength ) >= uxItemSize ) && 
        ( ( SIZE_MAX - sizeof( Queue_t ) ) >= ( size_t ) ( uxQueueLength * uxItemSize ) ) )
    {
        xQueueSizeInBytes = ( size_t ) ( uxQueueLength * uxItemSize );

        // 一次性申请“控制块 + 存储空间”。内存布局上：控制块在前，存储空间紧随其后
        pxNewQueue = ( Queue_t * ) pvPortMalloc( sizeof( Queue_t ) + xQueueSizeInBytes );

        if( pxNewQueue != NULL )
        {
            // 计算存储空间起始地址：跳过结构体本身的大小
            pucQueueStorage = ( ( uint8_t * ) pxNewQueue ) + sizeof( Queue_t );

            #if ( configSUPPORT_STATIC_ALLOCATION == 1 )
            {
                pxNewQueue->ucStaticallyAllocated = pdFALSE;
            }
            #endif

            // 初始化队列成员
            prvInitialiseNewQueue( uxQueueLength, uxItemSize, pucQueueStorage, ucQueueType, pxNewQueue );
        }
    }
    return pxNewQueue;
}

2.2 初始化与重置

初始化过程涉及指针的设定及链表的清空。

static void prvInitialiseNewQueue( const UBaseType_t uxQueueLength, const UBaseType_t uxItemSize, uint8_t * pucQueueStorage, const uint8_t ucQueueType, Queue_t * pxNewQueue )
{
    ( void ) ucQueueType;
    // 队列存储其实地址
    pxNewQueue->pcHead = ( int8_t * ) pucQueueStorage;
    // 队列保存最大元素个数
    pxNewQueue->uxLength = uxQueueLength;
    // 单个元素的大小
    pxNewQueue->uxItemSize = uxItemSize;
    
    ( void ) xQueueGenericReset( pxNewQueue, pdTRUE );
}

BaseType_t xQueueGenericReset( QueueHandle_t xQueue, BaseType_t xNewQueue )
{
    Queue_t * const pxQueue = xQueue;
    taskENTER_CRITICAL();
    {
        // 指向队列存储区的结束地址
        pxQueue->u.xQueue.pcTail = pxQueue->pcHead + ( pxQueue->uxLength * pxQueue->uxItemSize );
        // 初始化当前剩余元素为0
        pxQueue->uxMessagesWaiting = ( UBaseType_t ) 0U;
        // 写指针指向队列存储区的起始地址
        pxQueue->pcWriteTo = pxQueue->pcHead;
        
        /*
        这里的读取指针是指向读取位置的前一个位置 比如最旧的未读取数据在索引2位置
        读取指针指向1 主要是解决环形队列中满和空判断相同的问题 这里留了一个空位 
        判满和判空条件就不一样了
        */
        pxQueue->u.xQueue.pcReadFrom = pxQueue->pcHead + ( ( pxQueue->uxLength - 1U ) * pxQueue->uxItemSize );
        
        pxQueue->cRxLock = queueUNLOCKED;
        pxQueue->cTxLock = queueUNLOCKED;

        if( xNewQueue == pdFALSE ) // 如果是运行时重置，需处理可能阻塞的任务
        {
            if( listLIST_IS_EMPTY( &( pxQueue->xTasksWaitingToSend ) ) == pdFALSE )
            {
                // 队列清空了，唤醒优先级最高的发送阻塞任务
                if( xTaskRemoveFromEventList( &( pxQueue->xTasksWaitingToSend ) ) != pdFALSE )
                {
                    // 唤醒了高优先级任务，立即切换
                    queueYIELD_IF_USING_PREEMPTION();
                }
            }
        }
        else // 初始创建，仅需初始化链表
        {
            vListInitialise( &( pxQueue->xTasksWaitingToSend ) );
            vListInitialise( &( pxQueue->xTasksWaitingToReceive ) );
        }
    }
    taskEXIT_CRITICAL();
    return pdPASS;
}

3. 队列的发送机制

发送宏 xQueueSend 最终映射到 xQueueGenericSend。

#define xQueueSend( xQueue, pvItemToQueue, xTicksToWait ) \
    xQueueGenericSend( ( xQueue ), ( pvItemToQueue ), ( xTicksToWait ), queueSEND_TO_BACK )

3.1 发送核心逻辑 xQueueGenericSend

BaseType_t xQueueGenericSend( QueueHandle_t xQueue, const void * const pvItemToQueue, TickType_t xTicksToWait, const BaseType_t xCopyPosition )
{
    BaseType_t xEntryTimeSet = pdFALSE, xYieldRequired;
    TimeOut_t xTimeOut;
    Queue_t * const pxQueue = xQueue;

    for( ; ; ) // 核心死循环逻辑
    {
        taskENTER_CRITICAL();
        {
            // 1. 判断是否有空位（或者处于覆盖模式）
            if( ( pxQueue->uxMessagesWaiting < pxQueue->uxLength ) || ( xCopyPosition == queueOVERWRITE ) )
            {
                // 复制数据到队列
                xYieldRequired = prvCopyDataToQueue( pxQueue, pvItemToQueue, xCopyPosition );

                // 2. 检查是否有任务在因“空队列”而阻塞接收
                if( listLIST_IS_EMPTY( &( pxQueue->xTasksWaitingToReceive ) ) == pdFALSE )
                {
                    if( xTaskRemoveFromEventList( &( pxQueue->xTasksWaitingToReceive ) ) != pdFALSE )
                    {
                        // 唤醒了高优先级任务，立即切换
                        queueYIELD_IF_USING_PREEMPTION();
                    }
                }
                else if( xYieldRequired != pdFALSE )
                {
                    queueYIELD_IF_USING_PREEMPTION();
                }
                taskEXIT_CRITICAL();
                return pdPASS;
            }
            // 3. 队列已满：处理阻塞逻辑
            else
            {
                if( xTicksToWait == ( TickType_t ) 0 )
                {
                    taskEXIT_CRITICAL();
                    return errQUEUE_FULL;
                }
                else if( xEntryTimeSet == pdFALSE )
                {
                    vTaskInternalSetTimeOutState( &xTimeOut ); // 记录当前起始时间
                    xEntryTimeSet = pdTRUE;
                }
            }
        }
        taskEXIT_CRITICAL();

        // 挂起调度器并锁定队列，准备进入阻塞
        vTaskSuspendAll();
        prvLockQueue( pxQueue );

        if( xTaskCheckForTimeOut( &xTimeOut, &xTicksToWait ) == pdFALSE )
        {
            if( prvIsQueueFull( pxQueue ) != pdFALSE )
            {
                // 将任务同时挂载到：队列发送事件链表 + 全局延时链表
                vTaskPlaceOnEventList( &( pxQueue->xTasksWaitingToSend ), xTicksToWait );
                prvUnlockQueue( pxQueue );
                if( xTaskResumeAll() == pdFALSE )
                {
                    taskYIELD_WITHIN_API(); // 发起任务切换，任务在此“暂停”
                }
            }
            else
            {
                prvUnlockQueue( pxQueue );
                ( void ) xTaskResumeAll();
            }
        }
        else // 已超时
        {
            prvUnlockQueue( pxQueue );
            ( void ) xTaskResumeAll();
            return errQUEUE_FULL;
        }
    }
}

3.2 关于 for(;;) 循环的深度见解

非常重要： 这个函数有个比较奇怪的地方，不知道读者是否注意到，这个函数其实是包含在一个死循环中的，退出的地方有三个。第一是队列未满，复制成功；第二个是等待时间为0，直接退出，第三个是超时，直接退出。函数中调用了taskYIELD_WITHIN_API();宏定义，这个宏定义就是主动触发一次任务切换，因为在这之前，任务已经加入到延时/阻塞队列了，所以当前任务肯定无法继续执行了，就停在这里了。当这个队列有空位的时候，任务会被唤醒(稍后会详细分析)，唤醒之后，从这里启动，向下执行，但是这个是一个死循环，所有又会从头for循环的内容开始，检查队列是不是有空位，我们这里是假设了有空位了，所以会执行复制，然后退出。针对第三种退出条件就是超时，前面说过，如果超时时间不是portMAX_DELAY，那么就会进入到延时，当延时时间到了(这之前没有第一种情况发生)，这个任务也会恢复，然后也要从头开始执行判断是不是有空位，我们这里假设了没有空位，向下执行，判断是不是超时了，成立，退出。

理解上面这段话之后就可以知道，任务是怎么被队列阻塞，又是怎么从阻塞中恢复的了。这是站在任务的角度来看，我们以队列的视角来看，阻塞，恢复这个过程是怎么循环起来的。

任务向队列发送数据，判断有没有空位，如果有，看看队列是不是存在读取阻塞的任务，如果有恢复这个任务，读取队列时，会看看有没有发送阻塞的任务，如果有恢复这个任务。是不是，这样就形成了一个循环，在发送队列之后看看是不是有读取阻塞，在读取队列之后，看看是不是有发送阻塞。

当前上面这个循环是没有考虑超时的情况，如果是超时先到，那么这个任务就不是被队列唤醒了，而是直接调度器唤醒了。如果超时时间设置为portMAX_DELAY，这种情况最简单了，就是一直在全局挂起链表中，等待队列的唤醒。

4. 辅助函数分析

4.1 数据拷贝 prvCopyDataToQueue

此函数处理实际的内存拷贝，并根据位置决定是正常入队还是“插队”（发送到头部）。

static BaseType_t prvCopyDataToQueue( Queue_t * const pxQueue, const void * pvItemToQueue, const BaseType_t xPosition )
{
    BaseType_t xReturn = pdFALSE;
    UBaseType_t uxMessagesWaiting = pxQueue->uxMessagesWaiting;

    if( pxQueue->uxItemSize == ( UBaseType_t ) 0 ) // 信号量模式
    {
        #if ( configUSE_MUTEXES == 1 )
            if( pxQueue->uxQueueType == queueQUEUE_IS_MUTEX )
            {
                xReturn = xTaskPriorityDisinherit( pxQueue->u.xSemaphore.xMutexHolder );
                pxQueue->u.xSemaphore.xMutexHolder = NULL;
            }
        #endif
    }
    else if( xPosition == queueSEND_TO_BACK ) // 正常发送
    {
        ( void ) memcpy( ( void * ) pxQueue->pcWriteTo, pvItemToQueue, ( size_t ) pxQueue->uxItemSize );
        pxQueue->pcWriteTo += pxQueue->uxItemSize;
        if( pxQueue->pcWriteTo >= pxQueue->u.xQueue.pcTail )
        {
            pxQueue->pcWriteTo = pxQueue->pcHead;
        }
    }
    else // 发送到头部 (queueSEND_TO_FRONT / queueOVERWRITE) 很少用 不用太关心
    {
        ( void ) memcpy( ( void * ) pxQueue->u.xQueue.pcReadFrom, pvItemToQueue, ( size_t ) pxQueue->uxItemSize );
        pxQueue->u.xQueue.pcReadFrom -= pxQueue->uxItemSize;
        if( pxQueue->u.xQueue.pcReadFrom < pxQueue->pcHead )
        {
            pxQueue->u.xQueue.pcReadFrom = ( pxQueue->u.xQueue.pcTail - pxQueue->uxItemSize );
        }
    }

    pxQueue->uxMessagesWaiting = ( UBaseType_t ) ( uxMessagesWaiting + ( UBaseType_t ) 1 );
    return xReturn;
}

4.2 任务挂载 vTaskPlaceOnEventList

这是实现阻塞的核心：任务的“分身”被同时放入两个篮子。

void vTaskPlaceOnEventList( List_t * const pxEventList, const TickType_t xTicksToWait )
{
    // 1. 插入队列维护的事件链表（按任务优先级排序）
    vListInsert( pxEventList, &( pxCurrentTCB->xEventListItem ) );
    // 2. 插入全局延时/挂起链表 如果xTicksToWait=portMAX_DELAY，则插入到最挂起队列
    prvAddCurrentTaskToDelayedList( xTicksToWait, pdTRUE );
}

5. 总结

FreeRTOS 队列实现任务间同步和通信，底层本质上是通过精妙的链表操作实现的。通过将任务的“事件链表项”和“状态链表项”在不同链表间游走，内核实现了极高效率的任务管理。

队列的接收逻辑与发送逻辑基本是对称的。希望通过本篇分析，读者可以触类旁通。当你彻底掌握了这一机制，甚至可以尝试根据具体需求修改 FreeRTOS 源码或添加新的内核机制。

还有一个非常值得注意的问题是，队列经常会有一个ISR的版本API，核心区别就是中断没办法进行任务切换，并且绝对不可以阻塞，相当于xTicksToWait=0的效果，其实其他的都是差不多的，读者可以自己分析一下中断版本的函数。但是ISR版本中存在一个返回参数pxHigherPriorityTaskWoken因为任务本身无法进行任务切换，但是可以根据这个值的返回值，然后调用portYIELD_FROM_ISR(pxHigherPriorityTaskWoken)这里会根据结果触发一次pendsv中断进行任务切换。

声明： 本篇文章内容由作者原创撰写，并使用了 AI 进行辅助润色与排版优化，文章核心技术观点及逻辑内容已由作者本人严格审核。

# stm32 # 单片机
Logo
AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

cover

AtomGit 首发开源｜光轮智能发布全球首个十万小时级全模态人类行为开源数据集 EgoSuite-Open100K

avatar AtomGit开源社区
cover

AtomCode 全新上线万亿代码模型 LongCat-2.0。限时 7 天免费体验！

avatar AtomGit开源社区
cover

构筑赋能金融行业应用 Agent｜20 强决战北京，一场金融 Agent 的硬核答卷

avatar AtomGit开源社区