说实话,看到这篇的标题是“实践(3)”而不是“实践(1)”,我推测你已经跟着前两篇把FreeRTOS跑起来,也建过几个任务了,现在你多半是在纠结同一件事:任务建出来了,它们之间怎么“打交道”?这其实就是实时操作系统里最核心的一块:任务间通信与同步。裸机开发的时候,一段代码从头跑到尾,不存在两个逻辑同时抢资源的问题;但一旦上了RTOS,任务A在算数据,任务B在刷界面,任务C在等按键,彼此之间怎么传递信息、怎么协调先后、怎么保护共享变量,就成了决定系统能不能稳定跑起来的关键。
这篇实践笔记,我打算把FreeRTOS里最常用的几种通信方式一次性理清楚,包括队列、二值信号量、互斥量、计数信号量、任务通知和事件组。每个我都给实际代码,讲清楚参数怎么选、坑在哪里,最后用一个模拟项目把这些机制串起来跑通。适合正在从裸机转RTOS、或者已经建好任务但不知道怎么让它们协作的开发者参考。
1. 动手之前,先把几种通信方式的适用场景分清楚
我最早学FreeRTOS的时候,最大的困惑不是某个API不会用,而是不知道“什么时候该用哪个”。队列、信号量、互斥量、事件组、任务通知,功能上有重叠,新手很容易一锅粥。所以这次先不急着写代码,把每种机制的“性格”摸清楚,后面选型就顺了。
1.1 多任务系统里真正要解决的三个问题
任务之间要协作,归根结底就是三个问题:
第一个是数据传递。温度采集任务读到了ADC值,要交给显示任务刷新屏幕,这个值怎么过去?最直接的办法是搞一个全局变量,接收任务轮询检查,但这既浪费CPU,又容易产生读写竞争。
第二个是同步触发。按键扫描任务检测到一次短按,希望让LED闪烁任务立刻执行一次,而不是由后者自己傻等。这本质上是一个“你干完了叫我”的通知机制。
第三个是资源保护。多个任务都要往同一个串口打印日志,如果同时打印,字符就会交错。这时候需要的不是传递信息,而是让同一时刻只有一个任务能碰这个串口。
在FreeRTOS里,这三个问题分别对应了不同的机制,但是它们之间没有严格的一一对应关系。比如队列既能传数据,也能实现同步;信号量既能做互斥,也能做事件通知。正因为如此,选型才需要一点经验。
1.2 机制总览:一张表看懂每个工具的定位
我整理了一个简单的对照表,是我自己在项目里选型时常用的参考:
| 机制 | 核心用途 | 传递内容 | 典型场景 | 注意点 |
|---|---|---|---|---|
| 队列 | 数据传递 | 一段数据副本 | 采集任务给处理任务送数据 | 数据会被拷贝,不适合大块数据 |
| 二值信号量 | 同步 | 无 | 中断通知任务“事件发生了” | 没有数据内容,只有“有没有” |
| 计数信号量 | 资源计数 | 计数值 | 管理N个相同资源 | 计数值上限要定好 |
| 互斥量 | 资源独占 | 无 | 保护串口、LCD、Flash等共享外设 | 注意优先级反转问题 |
| 事件组 | 多条件组合 | 多个事件标志位 | 等待“温度到位”且“按键确认” | 事件标志位数量有限 |
| 任务通知 | 轻量同步/传值 | 一个32位值 | 一对一的快速通知 | 只能点对点,不能广播 |
看着这张表,你可能会想:队列啥都能干,是不是只用队列就行了?理论上确实可以,但实际工程里不推荐。队列要占内存,发送接收有拷贝开销,而且有些场景根本不需要传数据,只需要“踢一脚”,用信号量或者任务通知更轻量。选型的核心思路是:能用轻量的不用重量级的,能明确语义的不用模糊的。后面每一节我都会展开讲怎么选。
2. 队列实操:任务之间安全地“传纸条”
队列是FreeRTOS里用得最多的通信原语。它的本质是一个先进先出的缓冲区,发送方把数据拷贝进队尾,接收方从队头取出数据后,这段数据就被删除了。任务之间通过队列传递的数据是拷贝,不是指针,所以不用担心源数据后来被修改。
2.1 从裸机思维换成队列思维的三个关键参数
创建队列的API是xQueueCreate,只有两个参数:队列长度和单个数据项的大小。很多人在这里随便填,后面就出问题。
QueueHandle_t xAdcQueue; xAdcQueue = xQueueCreate(5, sizeof(uint16_t)); if (xAdcQueue == NULL) { // 创建失败,通常是堆内存不够 }第一个参数是队列深度,也就是最多能缓存多少个数据项。这个值取决于你的生产速度和消费速度的“水位差”。比如采集任务每10ms往队列里放一个数,处理任务要花50ms才能处理完一个数,那队列至少要能缓冲5个,不然处理任务一忙,后来的数据就无队可入,直接被丢掉。我习惯留至少两倍余量,防止突发峰值。
第二个参数是每个数据项占多少字节。这里有个特别容易踩的坑:如果你要传一个结构体,必须传sizeof(struct xxx),而不是sizeof(指针)。队列存的是数据的副本,不是指针本身。
typedef struct { uint16_t adc; uint32_t timestamp; } SensorData_t; // 错误写法:队列里实际只存了4字节指针 QueueHandle_t xQueue = xQueueCreate(5, sizeof(SensorData_t*)); // 正确写法:队列里存整个结构体副本 QueueHandle_t xQueue = xQueueCreate(5, sizeof(SensorData_t));我见过不止一个同学在这个地方栽跟头,队列创建得挺成功,发送也返回成功,但接收端拿到的数据永远是乱码。查了半天发现是队列存了指针大小,导致结构体只被拷贝了一部分。
2.2 发送端和接收端的标准写法
队列创建好之后,使用就分成两半:发送方和接收方。先看发送方的代码,这里以模拟读取ADC为例:
void vAdcTask(void *pvParameters) { uint16_t adcValue = 0; for (;;) { adcValue = readADC(); // 读取ADC原始值 // 如果队列满了,最多等100个tick if (xQueueSend(xAdcQueue, &adcValue, 100) != pdPASS) { // 发送失败,说明队列满了,可能要降低采集频率 } vTaskDelay(pdMS_TO_TICKS(10)); } }再看接收方:
void vProcessTask(void *pvParameters) { uint16_t receivedValue = 0; for (;;) { // portMAX_DELAY表示一直等,直到队列里有数据 if (xQueueReceive(xAdcQueue, &receivedValue, portMAX_DELAY) == pdPASS) { processValue(receivedValue); } } }这段代码里有几个细节值得说。发送方的等待时间我设了100个tick,也就是说队列满了我愿意等一小会儿,而不是一失败就丢数据。接收方用了portMAX_DELAY,配合当前场景很合理:处理任务没有别的事干,就在这蹲着等数据。如果接收任务还有其他周期性工作要干,就不应该死等,而是用有限超时,超时后去看看别的事。
2.3 中断里发数据,记住这套固定写法
队列最典型的应用场景之一,是从中断服务函数往里塞数据。但中断里的API和普通任务里的不一样,不能直接用xQueueSend,而是要用xQueueSendFromISR。
void EXTI_IRQHandler(void) { BaseType_t xHigherPriorityTaskWoken = pdFALSE; uint16_t keyCode = getKeyCode(); // 把按键值放入队列 xQueueSendFromISR(xKeyQueue, &keyCode, &xHigherPriorityTaskWoken); // 如果发送后发现有更高优先级的任务被唤醒,直接切换 portYIELD_FROM_ISR(xHigherPriorityTaskWoken); }这个xHigherPriorityTaskWoken变量相当关键。它的作用是告诉系统:“刚才这个操作唤醒了一个高优先级任务,现在应该立刻切换过去”。如果把它漏了,按键任务可能要等下一个tick中断才能被调度,响应时间会变得不稳定。我自己的习惯是:只要在中断里调用了FromISR后缀的API,就必须带上这个变量,并在最后调用portYIELD_FROM_ISR判断切换,不管当时想不想得起来原因。
提示:队列发送的中断版本不阻塞,没有超时参数。所以中断侧往队列塞数据时,如果队列满了,数据会被直接丢弃并返回
errQUEUE_FULL。在实时系统里,这通常意味着生产速度超过了消费速度,需要考虑加大队列深度或者提高接收任务优先级,而不是在中断里做处理。
3. 信号量与互斥量:资源保护不能靠运气
信号量和互斥量听名字很像,实际用起来也容易混淆。简单粗暴地区分:二值信号量是“事件通知”,互斥量是“锁”。二值信号量可以用于中断通知任务,但绝对不应该用来保护共享资源;互斥量专门用来保护共享资源,但绝对不要用在中断里。
3.1 二值信号量:中断和任务之间的“敲门铃”
二值信号量就像门口的一个铃铛,有人按一下,铃铛就响一次。任务在屋里等着,听到铃铛响就开门处理。它的特点是只记住“有没有发生过”,并不关心发生了多少次。如果一个中断在任务还没处理完之前又触发了第二次,信号量的计数值也不会变成2,依然只是“有事件”。
SemaphoreHandle_t xButtonSem; void vButtonTask(void *pvParameters) { for (;;) { // 等待按键中断给出信号量 if (xSemaphoreTake(xButtonSem, portMAX_DELAY) == pdPASS) { handleButtonPress(); } } } // 按键中断 void Button_ISR(void) { BaseType_t xHigherPriorityTaskWoken = pdFALSE; xSemaphoreGiveFromISR(xButtonSem, &xHigherPriorityTaskWoken); portYIELD_FROM_ISR(xHigherPriorityTaskWoken); }创建二值信号量用xSemaphoreCreateBinary()。这里有个隐藏知识点:刚创建的二值信号量是空的,第一次Take会阻塞,直到某个地方Give一次。很多新手在这里困惑“为什么我的任务卡住了”,就是因为创建完信号量忘了有人给它Give。
3.2 互斥量与优先级反转:一个真实的坑
互斥量和二值信号量的API长得几乎一样,一个是xSemaphoreCreateMutex,一个是xSemaphoreCreateBinary。但互斥量多了一个特性:优先级继承。这个特性专门用来解决一类叫“优先级反转”的问题。
我举一个具体的例子。系统里有三个任务:低优先级任务A、中优先级任务B、高优先级任务C。A先获得了一把互斥锁,正在写Flash。C这时候也想要这把锁,被迫阻塞等待。按道理A很快就能写完释放锁,C就能接着跑。但B这时候恰好就绪了,因为B的优先级比A高,系统立刻切到B执行。B一直占着CPU不放手,C虽然优先级最高,却因为A拿不到CPU而迟迟无法获得锁。这就是优先级反转:高优先级任务的执行时间被中优先级任务间接拖长了。
互斥量的优先级继承解决的就是这个场景:当C在等A手里的锁时,系统会把A的优先级临时提升到C的水平,让A能尽快抢回CPU执行完临界区,释放锁后再恢复原来的优先级。这样一来,B就没法在中间插队了。
SemaphoreHandle_t xUartMutex; void vLogTaskA(void *pvParameters) { for (;;) { xSemaphoreTake(xUartMutex, portMAX_DELAY); // 独占串口打印 printf("Task A log...\n"); xSemaphoreGive(xUartMutex); vTaskDelay(pdMS_TO_TICKS(50)); } }有人可能会问:临界区那么短,直接把中断关掉不就行了?关中断确实是保护共享资源的另一种手段,但它的粒度太粗,会影响系统的实时性。互斥量的价值在于,它只阻塞“也想访问这个资源的任务”,不阻塞其他无关任务,也不会把整个系统的中断都关掉。在嵌入式系统里,能用任务级阻塞解决的问题,尽量别用关中断来解决。
3.3 计数信号量:多资源场景的精确管理
计数信号量比二值信号量多了一个“数量”维度。比如你的系统里有3个DMA通道,4个任务都想用DMA传输数据,就可以创建一个最大计数值为3的计数信号量。任务使用前Take一个,用完Give一个,信号量内部的计数值保证了同时最多只有3个任务在用DMA。
SemaphoreHandle_t xDmaSem; void vTaskDemo(void *pvParameters) { for (;;) { // 获取一个DMA通道 xSemaphoreTake(xDmaSem, portMAX_DELAY); useDmaChannel(); // 释放 xSemaphoreGive(xDmaSem); } }创建时用xSemaphoreCreateCounting(3, 3),第一个参数是最大计数值,第二个是初始计数值。初始值设成3表示3个资源一开始都是可用的;如果是外部设备发来的事件计数,初始值往往设成0,表示“还没有事件到达”。计数信号量的好处是:即使有突发情况导致事件堆积,计数值也能慢慢累计,不会像二值信号量那样丢事件。
4. 任务通知和事件组:两种更轻量的通信思路
如果说队列、信号量是FreeRTOS的“标准配置”,那么任务通知就是近几年特别受欢迎的高性能方案,而事件组则是处理复杂条件组合的好帮手,两者各有各的绝活。
4.1 任务通知:一对一的轻量同步
任务通知的核心思想是:每个任务内部内置了一个32位的通知值,其他任务可以直接修改这个值来唤醒它。不需要创建单独的队列或者信号量对象,省掉了一整块内核对象的开销,速度也比信号量快不少。
TaskHandle_t xTaskHandleProcess; // 任务A:发通知 void vAdcTask(void *pvParameters) { for (;;) { readAdcAndUpdate(); // 给处理任务发一个通知,通知值加1 xTaskNotifyGive(xTaskHandleProcess); vTaskDelay(pdMS_TO_TICKS(10)); } } // 任务B:收通知 void vProcessTask(void *pvParameters) { uint32_t notifyCount; for (;;) { // 等待通知,pdTRUE表示进入阻塞前清空计数 notifyCount = ulTaskNotifyTake(pdTRUE, portMAX_DELAY); if (notifyCount > 1) { // 说明积压了多个通知,可以考虑批量处理 } processLatestData(); } }这个模式看起来和二值信号量很像,但有个关键区别:任务通知有“计数”能力。ulTaskNotifyTake返回值就是通知累加的次数,也就是说如果发送方连续给了10次通知,接收方醒来后拿到的是10,而不是“丢失”事件。这在采集场景特别好用,可以用一次处理把积累的数据全部消化掉。
除了xTaskNotifyGive,任务通知还有更灵活的版本xTaskNotify,可以设置具体的事件值,甚至带上一个32位的数据,相当于一个极简版队列。但要注意,它只能点对点通信:一个任务只能被另一个任务通知,无法实现“一个通知唤醒多个任务”的广播效果。
4.2 事件组:多条件组合触发的场景
有时候一个任务不是等单个事件,而是要等“多个条件同时满足”或者“任意一个条件满足”。比如设备要开始打印,必须同时满足“纸张已装好”“墨盒已安装”“盖子已合上”三个条件。如果用信号量硬凑,代码会非常别扭。事件组就是专门解决这个问题的。
EventGroupHandle_t xEventGroup; #define BIT_PAPER_READY (1 << 0) #define BIT_INK_READY (1 << 1) #define BIT_COVER_CLOSED (1 << 2) void vPrinterTask(void *pvParameters) { EventBits_t bits; for (;;) { // 等待三个条件都满足 bits = xEventGroupWaitBits( xEventGroup, BIT_PAPER_READY | BIT_INK_READY | BIT_COVER_CLOSED, pdTRUE, // 满足后自动清除这些位 pdTRUE, // 要等所有位都满足才返回 portMAX_DELAY ); startPrinting(); } } // 其他任务或者中断里设置各自的条件 void vSensorTask(void *pvParameters) { // 检测纸张 if (isPaperReady()) { xEventGroupSetBits(xEventGroup, BIT_PAPER_READY); } }第四个参数pdTRUE表示“逻辑与”,等所有位都置1才返回;pdFALSE表示“逻辑或”,任何一个位置1就返回。第五个参数是超时。这个API的返回值可以用来判断到底是因为哪些位满足而退出的,从而区分不同的处理路径。
事件组还有一个独特的优势:它可以被多个任务同时等待,也就是说可以实现“广播”效果,这是任务通知做不到的。
4.3 什么时候别用任务通知
任务通知虽然快,但有两个限制比较明显。第一个是只能一对一,不能广播。第二个是,接收方如果没有进入等待状态,通知值就只是一个数字,不像信号量那样有“等待队列”的概念。如果你有多个任务在同一个信号量上阻塞,Give唤醒的是优先级最高的那个;但任务通知没有这个能力,发送方只能指定给某一个任务。
所以我的经验是:中断到任务的通知,如果只有一个任务接收,优先用任务通知;如果可能有多个任务都在等待同一个事件,就用二值信号量或事件组。这个判断标准基本能覆盖绝大多数应用场景。
5. 模拟项目X:把通信机制串成一条完整流水线
前面讲了各种机制,但说实话分开看容易都会,合起来就懵。这里我用一个模拟项目X把前面所有内容串起来跑一遍。这个项目的需求是:采集环境温湿度数据,经过滤波处理后,将结果显示到屏幕,同时支持按键切换显示模式。
5.1 整体流程与任务设计
系统拆成三个任务加一个中断:
| 角色 | 优先级 | 职责 |
|---|---|---|
| 数据采集任务 | 2 | 每100ms读取一次传感器,把原始数据放入队列 |
| 数据处理任务 | 3 | 从队列取数据,做滑动平均滤波,更新显示缓冲区 |
| 按键处理任务 | 1 | 等待按键中断发来的通知,切换显示模式 |
| 定时器中断 | 最高 | 定期触发传感器读取 |
数据流是:定时器中断触发采集任务读传感器,采集任务把数据通过队列交给处理任务,处理任务计算完更新显示。按键中断通过任务通知直接唤醒按键处理任务。
创建各个元素的代码放在一个初始化函数里:
QueueHandle_t xSensorQueue; SemaphoreHandle_t xUartMutex; void vAppInit(void) { // 队列存放结构体 xSensorQueue = xQueueCreate(10, sizeof(SensorData_t)); // 互斥量保护串口 xUartMutex = xSemaphoreCreateMutex(); // 创建三个任务,句柄保存下来 xTaskCreate(vSensorTask, "Sensor", 256, NULL, 2, NULL); xTaskCreate(vProcessTask, "Process", 512, NULL, 3, NULL); xTaskCreate(vKeyTask, "Key", 128, NULL, 1, NULL); // 启动调度器 vTaskStartScheduler(); }注意处理任务的栈大小我给了512个word,比另外两个大不少,因为它要做滤波计算,会有局部数组。栈大小不够是RTOS系统最常见的崩溃原因之一,后面会专门说这个问题。
5.2 采集任务与处理任务的完整代码
采集任务的代码重点在于“定时读取”和“入队”:
void vSensorTask(void *pvParameters) { SensorData_t data; TickType_t lastWakeTime = xTaskGetTickCount(); for (;;) { vTaskDelayUntil(&lastWakeTime, pdMS_TO_TICKS(100)); data.adc = readSensor(); data.timestamp = xTaskGetTickCount(); // 入队,如果满了就丢弃这次数据,但记录错误 if (xQueueSend(xSensorQueue, &data, 0) != pdPASS) { logError("Queue full"); } } }这里用vTaskDelayUntil保证采集周期严格100ms,不受代码执行时间波动影响。它跟vTaskDelay的区别在于:vTaskDelay是从调用时开始延时,如果任务执行本身花了20ms,实际周期就是120ms;而vTaskDelayUntil是以上次唤醒时间为基准计算,周期严格固定在100ms。
处理任务这边,我用了滑动平均滤波,每收到5个数据求一次平均更新显示:
void vProcessTask(void *pvParameters) { SensorData_t data; uint32_t sum = 0; uint8_t count = 0; for (;;) { // 等数据,没有就挂着 xQueueReceive(xSensorQueue, &data, portMAX_DELAY); sum += data.adc; count++; if (count >= 5) { uint16_t avg = sum / 5; // 更新显示缓冲区时用互斥量保护 xSemaphoreTake(xUartMutex, portMAX_DELAY); updateDisplayBuffer(avg); xSemaphoreGive(xUartMutex); // 重置累计 sum = 0; count = 0; } } }按键处理任务用任务通知接收中断信号,这个已经在前面讲过了,这里不重复贴代码。整个项目的关键点在于:数据传递用队列,显示缓冲区保护用互斥量,按键通知用任务通知,每个机制都放在了自己最合适的场景。
5.3 实测过程中发现的两个问题
这个项目我实际跑过,遇到了两个有代表性的问题。
第一个问题是采集任务优先级设为2,处理任务优先级设为3,结果处理任务一忙起来,采集任务偶尔会错过vTaskDelayUntil的唤醒点,导致采集周期抖动。原因在于处理任务优先级更高,而采集任务每次被抢占后,vTaskDelayUntil计算下次唤醒时间是按绝对时间算的,一旦错过就会补偿性地立即执行,看起来就是周期突然短了一截。排查后发现,处理任务虽然优先级高,但它的工作其实很快,只是每次都要等队列数据,不会真的长时间霸占CPU。把两个任务优先级对调,也就是采集任务2改成3、处理任务3改成2之后,问题消失。
第二个问题比较隐蔽:有时候串口打印会出现乱码,而且只在处理任务特别忙的时候出现。排查了一圈,发现是显示缓冲区的更新和打印任务用了不同的互斥量,等于两个锁保护同一个资源,自然等于没锁。统一改成用同一个互斥量、或者把这部分访问集中到一个任务里以后,乱码就没了。这个经验就是:同一个资源必须严格对应同一把锁,别想着“大概没冲突”。
6. 常见问题与排查技巧实录
把我在实际项目里遇到的高频问题整理成一张速查表,一部分是网上常问的,一部分是自己踩过的。建议先收藏,出了问题回来查。
6.1 五个高频问题速查
| 现象 | 可能原因 | 排查思路 |
|---|---|---|
| 任务创建后从不运行 | 栈大小不足,任务创建失败 | 检查xTaskCreate返回值是否为pdPASS |
| 系统卡死,调试器停在HardFault | 栈溢出或非法内存访问 | 检查任务栈大小,启用vApplicationStackOverflowHook钩子函数 |
| 队列发送返回失败 | 队列满了 | 增大队列深度,或者降低生产频率 |
| 中断后任务不响应 | 忘记调用portYIELD_FROM_ISR | 检查中断服务函数是否有唤醒动作 |
| 多个任务同时打印乱码 | 共享资源未保护 | 给串口加互斥量,或用专门任务统一打印 |
栈溢出这个问题值得多说两句。FreeRTOS提供了栈溢出检测机制,可以通过configCHECK_FOR_STACK_OVERFLOW配置开启。开启后如果发生溢出,系统会调用vApplicationStackOverflowHook钩子函数,你可以在这里设置断点或者点亮一个错误LED。但要注意,这个检测不是100%实时,尤其在中断嵌套较深的情况下可能来不及触发。我自己在调试时还会额外用uxTaskGetStackHighWaterMark来查看任务剩余栈空间:
UBaseType_t freeStack = uxTaskGetStackHighWaterMark(xSensorTaskHandle); // freeStack越小,说明这个任务离栈溢出越近这个函数的返回值是任务创建以来看到的“最低剩余栈空间”,不是当前值。我发现一个任务如果剩余栈空间低于100个word,就基本可以断定栈不够了,要赶紧加大。
6.2 排查工具的坑与心得
市面上不少调试器支持直接查看RTOS内核对象,比如队列的使用情况、任务的阻塞状态。有条件的话尽量用上,比纯靠猜效率高太多。
没有调试工具也没关系,我分享一个土办法:在关键节点用串口打印标记,但注意别在中断里直接打印。中断里做打印会拖慢中断响应,甚至引发不可预期的问题。正确的做法是在中断里置一个标志位,然后在任务里轮询这个标志再打印。另外,如果怀疑是任务饿死或者优先级配置问题,可以临时把所有任务的优先级都设成相同值,看系统是否还能跑,如果相同优先级时正常、不同优先级时异常,那问题多半出在优先级设计上。
说到优先级,我的经验是:优先级不要超过5个档次。很多新手喜欢把每个任务都设成不同的优先级,结果系统调度逻辑复杂到无法推理。优先级的差距只要能让紧急任务抢到CPU就够了,没必要分太细。中断里的逻辑能短则短,把耗时操作全部放到任务里。
写在自己项目里的一段话
这篇写到这里,基本上把我平时做FreeRTOS项目时最常用的通信机制都盘了一遍。如果你正准备往自己的项目里加通信逻辑,我的建议是先画出数据流图,标清楚谁在产生数据、谁在消费数据、谁在保护共享资源,然后再去套机制,而不是一上来就想“我要用队列还是信号量”。我在实际使用中最大的体会是,FreeRTOS的这些通信原语其实都是好工具,但真正让系统稳定的是对任务划分、优先级分配和资源归属的清晰认知。工具选错可以换,但架构乱了,后面每一步都会很难受。