os操作系统——第5讲：任务间的鹊桥

Nice__J

334人浏览 · 2026-06-17 10:51:08

Nice__J · 2026-06-17 10:51:08 发布

第5讲：任务间的鹊桥——队列、信号量与互斥量

上一讲我们让任务学会了“睡觉”（阻塞延时），CPU 不再空转。但任务之间依然是孤岛：
传感器任务采集到了数据，怎么送给处理任务？
两个任务同时想修改同一块内存，除了粗暴关中断，有没有更优雅的“约法三章”？
一个低优先级任务拿了锁不放，高优先级任务却干等着——这种“优先级反转”如何化解？

这一讲，我们将搭建任务间的“鹊桥”：信号量（同步）、互斥量（锁）、消息队列（数据传输）。从此，任务可以协作、通信、互斥，真正成为有机的整体。

1. 从“关中断”到“同步原语”

前几讲我们用 enter_critical()/exit_critical() 解决了共享变量的互斥问题。但这把“大锁”有两个硬伤：

屏蔽一切中断：关中断期间，不仅其他任务无法运行，连高优先级的中断服务都被延迟，不适合保护长时间的临界区。
无法携带信息：关中断只能保证“独占”，但任务之间还需要传递“数据已经就绪”这类事件信号。

于是，操作系统提供了一系列同步与通信原语：

信号量（Semaphore）：用于任务同步（事件通知）或资源计数（例如一个池子里有 N 个可用资源）。
互斥量（Mutex）：专门用于互斥访问，解决了优先级反转问题。
队列（Queue）：任务间传递数据（消息），同时自带同步（队列空时接收方阻塞，队列满时发送方阻塞）。

2. 信号量：从“事件旗语”到“资源计数”

信号量可以看作一个整数计数器，配合两个原子操作：take（P 操作，减一）和 give（V 操作，加一）。
计数器的含义取决于应用场景：

二值信号量（0/1）：用于任务间同步。例如，中断发生后 give，等待的任务 take 成功，就会从阻塞中醒来。
计数信号量：表示可用资源的数量。例如，一个内存池有 5 个 buffer，初始值 5。任务需要 buffer 时 take（减少 1），用完 give（增加 1）。

2.1 信号量的数据结构

typedef struct semaphore {
    int count;                 // 计数器
    tcb_t *wait_list;          // 等待这个信号量的任务链表
} sem_t;

2.2 核心操作

void sem_init(sem_t *sem, int init_count) {
    sem->count = init_count;
    sem->wait_list = NULL;
}

void sem_take(sem_t *sem) {
    enter_critical();
    if (sem->count > 0) {
        sem->count--;
        exit_critical();
        return;   // 有资源，直接成功
    }
    // 没有资源，阻塞当前任务
    current_task->state = TASK_BLOCKED;
    // 将当前任务挂入 sem->wait_list
    append_to_wait_list(&sem->wait_list, current_task);
    exit_critical();
    schedule();   // 让出 CPU
    // 注意：当任务被唤醒重新运行时，已经成功 take（由 give 操作保证了 count 已被减）
}

void sem_give(sem_t *sem) {
    enter_critical();
    if (sem->wait_list != NULL) {
        // 有任务在等待，唤醒第一个任务（不增加 count，直接移交资源）
        tcb_t *task = remove_first_from_wait_list(&sem->wait_list);
        task->state = TASK_READY;
        insert_into_ready_list(task);
    } else {
        sem->count++;   // 无等待者，增加可用资源数
    }
    exit_critical();
    // 注意：这里不直接 schedule，让当前任务继续，等待调度点（如 Tick 中断）再切换
}

为什么 give 时不直接调度？ 如果当前任务优先级很低，而刚唤醒的任务优先级很高，理论上应该立即抢占。但为了简单，可以在 give 后调用 schedule() 检查是否需要抢占。真正的 RTOS 会在释放信号量时检查等待队列中是否有更高优先级的任务，如果有则主动抢占。

2.3 典型用法：中断与任务同步

sem_t rx_sem;

void UART_IRQHandler(void) {
    char c = UART->DR;
    buffer[head] = c;    // 存入环形缓冲区
    sem_give(&rx_sem);    // 通知处理任务
}

void uart_task(void) {
    while (1) {
        sem_take(&rx_sem);   // 等待数据到达
        // 从缓冲区取出处理
    }
}

这样，处理任务在没有数据时完全阻塞，不消耗 CPU；中断到来时快速 give，处理任务立即被唤醒。

3. 互斥量：为了解决“优先级反转”

信号量也可以用于互斥（初始化 count=1，take 加锁，give 解锁）。但这会带来一个经典问题：优先级反转。

3.1 优先级反转演示

假设系统有三个任务：

H：高优先级
M：中优先级
L：低优先级
共享一个资源，通过二值信号量保护。

L 获取信号量，进入临界区。
H 抢占 L（因为 H 优先级高），试图获取信号量 → 被阻塞（因为 L 持有锁）。
M 中优先级任务没有使用该资源，但它优先级高于 L，所以 M 开始运行，占用了 CPU。
结果：H 被 M 间接阻塞了，尽管 M 根本不需要那个锁。H 反而要等 M 执行完，L 才能继续执行并释放锁。

这就是优先级反转——高优先级任务被不相干的中优先级任务拖延，严重破坏实时性。

3.2 互斥量的解决方案：优先级继承

互斥量（Mutex）提供 优先级继承 协议：

当低优先级任务 L 持有互斥量，而高优先级任务 H 试图获取时，系统会临时将 L 的优先级提升到 H 的级别。
这样 L 就可以不被 M 抢占，快速执行完临界区并释放互斥量。
释放后，L 的优先级恢复原状。

代码实现中，互斥量比信号量多记录一个 holder（当前持有任务）和 original_priority。

typedef struct mutex {
    int locked;
    tcb_t *holder;             // 当前持有锁的任务
    int original_priority;     // 用于继承（暂存）
    tcb_t *wait_list;
} mutex_t;

mutex_take 的逻辑会检查：如果锁已被占用，且当前请求者的优先级高于持有者的优先级，则提升持有者的优先级（继承）。
实现细节稍复杂，但原理清晰。

注意：优先级继承只能减轻反转，不能完全消除。如需彻底解决，需要“优先级天花板协议”或“随机提升”。但大多数 RTOS 的互斥量都实现了优先级继承。

4. 消息队列：带缓冲的任务邮局

有了信号量，我们可以同步数据就绪事件，但数据本身还在共享缓冲区里，仍需额外保护。
消息队列把数据和同步打包在一起：发送方把数据复制到队列内部，接收方从队列取出。队列满时发送方可阻塞（可选），队列空时接收方阻塞。

4.1 数据结构

#define QUEUE_MSG_SIZE 4   // 每个消息4字节（例如一个指针或整数值）
#define QUEUE_LEN    10

typedef struct queue {
    char buffer[QUEUE_LEN][QUEUE_MSG_SIZE];
    int head;               // 写索引
    int tail;               // 读索引
    int count;              // 当前消息数
    sem_t mutex;            // 保护内部结构的二值信号量
    sem_t available;        // 计数信号量，表示消息数量（用于接收）
    sem_t free_slots;       // 计数信号量，表示空位（用于发送）
} queue_t;

4.2 队列操作

void queue_send(queue_t *q, void *msg) {
    sem_take(&q->free_slots);   // 等待有空位
    sem_take(&q->mutex);        // 互斥访问队列指针
    
    // 拷贝消息到 buffer[head]
    memcpy(q->buffer[q->head], msg, QUEUE_MSG_SIZE);
    q->head = (q->head + 1) % QUEUE_LEN;
    q->count++;
    
    sem_give(&q->mutex);
    sem_give(&q->available);    // 通知有新消息
}

void queue_recv(queue_t *q, void *out) {
    sem_take(&q->available);    // 等待有消息
    sem_take(&q->mutex);
    
    memcpy(out, q->buffer[q->tail], QUEUE_MSG_SIZE);
    q->tail = (q->tail + 1) % QUEUE_LEN;
    q->count--;
    
    sem_give(&q->mutex);
    sem_give(&q->free_slots);   // 空位增加
}

这种实现方式避免了显式的关中断，只使用信号量，代码清晰，且可扩展至多核（只要信号量本身支持多核，第8讲会进入多核锁）。

5. 优先级反转的真实案例与教训

1997年，美国火星探路者号着陆火星后，突然发生系统复位，导致数据丢失。
调查发现：系统中有一个低优先级的“气象数据收集任务”持有了互斥量，被一个中优先级的“通信任务”抢占，而高优先级的“总线管理任务”等待互斥量——优先级反转！
系统设计时没有使用优先级继承的互斥量，而是用了普通信号量。
最终，火星车通过远程补丁，启用了优先级继承，问题解决。

这个故事告诉我们：互斥量和信号量绝不能混用。

6. 本讲小结 & 下集预告

今天我们搭建了三座鹊桥：

信号量：计数资源或事件通知，适合生产者-消费者模型、中断同步。
互斥量：带优先级继承的互斥锁，解决优先级反转，保护共享资源。
消息队列：数据传递 + 自动同步，是任务间通信的最常用手段。

有了这些原语，我们的小操作系统已经足够应对很多嵌入式场景。但还有一个更高级的概念尚未触及：如何让普通任务也能安全地调用内核服务？ 比如一个用户任务申请内存、发送队列——如果中途被中断打断会如何？

下一讲：函数调用也是任务——软件中断与系统调用
我们将引入 SVC 软中断，区分用户态和内核态，为操作系统的安全性和完整性打下基础。

✍️ 思考与练习

模拟反转：用你熟悉的 RTOS（如 FreeRTOS）写一个优先级反转的 demo：三个任务 H、M、L，L 持有锁，M 死循环，H 尝试获取锁。观察 H 的运行延迟。然后把二值信号量换成互斥量，对比现象。
实现轻量队列：在不使用信号量的情况下，用关中断实现一个环形队列的 send/recv，比较两种方法的代码复杂度和性能。
思考死锁：如果任务 A 持有锁 1，等待锁 2；任务 B 持有锁 2，等待锁 1。会发生什么？操作系统如何检测和避免死锁？
阅读源码：查找 Zephyr 或 FreeRTOS 中 xSemaphoreTake 和 xQueueSend 的实现，注意其临界区处理方式。

欢迎在评论区分享你遇到过的优先级反转案例或队列设计的心得！