知识门户

返回

第 10 章：任务间通信——Channel、Signal、Mutex 与 PubSub

Part2 深入embassy框架

第 10 章：任务间通信——Channel、Signal、Mutex 与 PubSub#

本章要回答的问题：Embassy 的任务之间如何传递数据？Channel、Signal、Mutex、PubSub 分别适用于什么场景？与 FreeRTOS 的队列/信号量/事件组有何对应关系？

本章定位：掌握 Embassy 的四种同步原语。每种原语都有完整的 API 讲解、C/FreeRTOS 对比、使用场景分析和实战示例。学完本章，你将能够为任何多任务通信需求选择正确的原语。

10.1 问题：任务之间如何交换数据？#

10.1.1 C 世界的”三件套”#

在 C + FreeRTOS 中，任务间通信依赖三个机制：

// 1. 队列（Queue）：传递数据
QueueHandle_t sensor_queue = xQueueCreate(4, sizeof(SensorData));
xQueueSend(sensor_queue, &data, portMAX_DELAY);
xQueueReceive(sensor_queue, &data, portMAX_DELAY);

// 2. 信号量（Semaphore）：同步/通知
SemaphoreHandle_t data_ready = xSemaphoreCreateBinary();
xSemaphoreGive(data_ready);
xSemaphoreTake(data_ready, portMAX_DELAY);

// 3. 互斥量（Mutex）：保护共享资源
SemaphoreHandle_t i2c_mutex = xSemaphoreCreateMutex();
xSemaphoreTake(i2c_mutex, portMAX_DELAY);
// 使用 I2C...
xSemaphoreGive(i2c_mutex);
c

C 的问题：

问题示例
类型不安全xQueueSend(queue, &data, ...) 中 &data 是 void*，传错类型编译器不管
大小不匹配队列创建时 sizeof(SensorData) 与发送时的结构体大小不一致 → 内存损坏
死锁风险两个任务以不同顺序获取两个 Mutex → 死锁（运行时才发现）
优先级反转低优先级任务持有 Mutex，高优先级任务被阻塞 → 需要优先级继承
忘记释放xSemaphoreTake 后忘记 xSemaphoreGive → 其他任务永久阻塞
全局变量不用 RTOS 原语时，volatile 全局变量 + 临界区 → 竞态条件

10.1.2 Embassy 的答案#

Embassy 提供四种通信原语，全部在 embassy-sync crate 中：

Embassy 原语功能FreeRTOS 对应C 裸机对应
Channel多对多数据传递Queue环形缓冲区 + 标志
Signal一对一单值通知Binary Semaphorevolatile 标志
Mutex互斥访问共享资源Mutex Semaphore临界区 + 标志
PubSub一对多广播Event Group + Queue回调函数列表

Embassy 的核心优势：

  1. 类型安全：Channel<Mutex, SensorData, 4> 在编译期确定传递的数据类型
  2. 无死锁：协作式调度 + 单线程 Executor → 不可能死锁
  3. 无优先级反转：没有优先级 → 没有反转
  4. 无 volatile：借用检查器保证数据一致性
  5. 零堆分配：所有原语都是 static 变量，编译期确定大小

10.2 Channel：多对多数据传递#

10.2.1 什么是 Channel？#

Channel 是一个固定容量的异步 FIFO 队列：

  • 发送方：send().await——队列满时挂起
  • 接收方：receive().await——队列空时挂起
  • 容量：编译期确定（泛型参数）
  • 多对多：多个任务可以同时发送和接收
┌──────────┐     send().await      ┌─────────────────┐     receive().await    ┌──────────┐
│  Task A  │ ──────────────────►   │                 │  ◄────────────────────  │  Task X  │
│ (生产者) │                       │    Channel      │                         │ (消费者) │
└──────────┘                       │  [T; N]         │                         └──────────┘
                                   │                 │
┌──────────┐     send().await      │  ┌───┬───┬───┐ │     receive().await    ┌──────────┐
│  Task B  │ ──────────────────►   │  │ 0 │ 1 │ 2 │ │  ◄────────────────────  │  Task Y  │
│ (生产者) │                       │  └───┴───┴───┘ │                         │ (消费者) │
└──────────┘                       │   容量 = N      │                         └──────────┘
                                   └─────────────────┘
plaintext

10.2.2 API 详解#

use embassy_sync::channel::Channel;
use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex;

// 声明：Channel<互斥锁类型, 数据类型, 容量>
static SENSOR_CH: Channel = Channel::new();
//                        ↑                      ↑            ↑
//                  保护内部状态的锁        传递的数据类型    队列容量（编译期）

#[derive(Clone, Copy)]
struct SensorData {
    temperature: i16,
    humidity: u16,
    timestamp: u32,
}
rust

发送操作#

接收操作#

// 异步接收（队列空时挂起）
let data: SensorData = SENSOR_CH.receive().await;

// 尝试接收（不挂起，立即返回）
match SENSOR_CH.try_receive() {
    Ok(data) => { /* 收到数据 */ }
    Err(TryReceiveError::Empty) => { /* 队列为空 */ }
}
rust

查询状态#

// 当前队列中的元素数量
let len: usize = SENSOR_CH.len();

// 队列是否为空/满
let is_empty: bool = SENSOR_CH.is_empty();
let is_full: bool = SENSOR_CH.is_full();

// 剩余容量
let capacity: usize = SENSOR_CH.capacity();  // 始终为 4（编译期确定）
rust

10.2.3 完整示例：传感器 → 处理 → 显示#

10.2.4 与 FreeRTOS Queue 的对比#

方面FreeRTOS xQueueEmbassy Channel
创建xQueueCreate(4, sizeof(SensorData))static CH: Channel<..., SensorData, 4> = Channel::new()
类型安全❌ void*（传错类型无警告）✅ 泛型参数（编译期检查）
内存分配动态（pvPortMalloc）静态（.bss 段）
发送xQueueSend(q, &data, timeout)CH.send(data).await
接收xQueueReceive(q, &data, timeout)let data = CH.receive().await
超时pdMS_TO_TICKS(100)select(send, Timer::after(...))
满/空检查uxQueueSpacesAvailable()CH.is_full() / CH.is_empty()
中断安全xQueueSendFromISR()CH.try_send()（在中断中）
所有权复制（memcpy）移动（零拷贝）
数据大小限制无（但大对象复制开销大）建议 Copy 类型（小对象）

关键差异：所有权 vs 复制

// FreeRTOS：数据被复制到队列中
SensorData data = { .temp = 250, .hum = 600 };
xQueueSend(queue, &data, 0);  // memcpy 到队列内部缓冲区
// data 仍然有效，可以继续使用
data.temp = 300;  // 不影响队列中的副本
c
// Embassy：数据被移动到 Channel 中
let data = SensorData { temperature: 250, humidity: 600 };
SENSOR_CH.send(data).await;
// data 已经被移动！不能再使用
// data.temperature = 300;  // ❌ 编译错误：use of moved value
rust

对于 Copy 类型（如 SensorData），移动和复制的机器码完全相同。但对于大对象，Rust 的所有权系统强制你思考”谁拥有这个数据”——这避免了 C 中常见的”谁负责释放”问题。

10.2.5 Channel 容量选择指南#

容量适用场景行为
1只关心最新值（如传感器读数）生产者可能被阻塞（队列满）
2-4允许短暂突发（如 UART 字节流）缓冲突发，平滑速率差异
8-16速率差异大（如高速 ADC → 慢速处理）较大缓冲，但占用更多 RAM
> 16极少需要（考虑是否设计有问题）RAM 开销显著

RAM 计算：

Channel 占用 = 容量 × sizeof(T) + 内部状态（~16 字节）

示例：Channel<..., SensorData, 4>
     = 4 × 8 字节 + 16 字节
     = 48 字节

对比 FreeRTOS Queue：
     = 4 × 8 字节 + 队列头（~80 字节）+ 动态分配开销
     = ~112 字节 + malloc 碎片
plaintext

10.3 Signal：一对一单值通知#

10.3.1 什么是 Signal？#

Signal 是一个容量为 1 的 Channel——但它有特殊的覆盖语义：

  • 如果 Signal 为空：signal(value) 存储值
  • 如果 Signal 已有值：signal(new_value) 覆盖旧值
  • wait().await：取出值（Signal 变空），如果为空则挂起
┌──────────┐     signal(42)      ┌─────────────┐     wait().await    ┌──────────┐
│  Task A  │ ─────────────────►  │   Signal    │  ◄────────────────  │  Task B  │
│ (发送方) │                     │  [42]       │                     │ (接收方) │
└──────────┘                     │  容量 = 1   │                     └──────────┘
                                 └─────────────┘

如果 Task A 连续发送：
  signal(1) → Signal = [1]
  signal(2) → Signal = [2]  ← 覆盖了 1！
  signal(3) → Signal = [3]  ← 覆盖了 2！

Task B 的 wait() 只会收到 3（最新值）
plaintext

10.3.2 API 详解#

use embassy_sync::signal::Signal;
use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex;

static BUTTON_SIGNAL: Signal = Signal::new();

#[derive(Clone, Copy)]
enum ButtonEvent {
    Pressed,
    Released,
    LongPress,
}
rust

发送（通知）#

// 发送信号（永不阻塞，永不失败）
BUTTON_SIGNAL.signal(ButtonEvent::Pressed);

// 在中断中发送（安全！）
#[interrupt]
fn EXTI0() {
    BUTTON_SIGNAL.signal(ButtonEvent::Pressed);
    // 清除中断标志...
}
rust

接收（等待）#

// 异步等待信号（Signal 为空时挂起）
let event: ButtonEvent = BUTTON_SIGNAL.wait().await;

// 非阻塞检查
match BUTTON_SIGNAL.try_take() {
    Some(event) => { /* 有信号 */ }
    None => { /* 无信号 */ }
}
rust

10.3.3 完整示例：按键事件通知#

10.3.4 与 FreeRTOS Binary Semaphore 的对比#

方面FreeRTOS Binary SemaphoreEmbassy Signal
创建xSemaphoreCreateBinary()static SIG: Signal<..., T> = Signal::new()
发送xSemaphoreGive(sem)SIG.signal(value)
接收xSemaphoreTake(sem, timeout)let v = SIG.wait().await
携带数据❌ 不能（只是”有/无”）✅ 可以（泛型类型 T）
覆盖语义多次 Give = 一次（二值）多次 signal = 保留最新值
中断安全xSemaphoreGiveFromISR()SIG.signal()（直接调用，安全）
内存动态分配静态分配
类型安全❌ 无数据✅ 编译期类型检查

Signal 的最大优势：它可以携带类型化的数据。FreeRTOS 的 Binary Semaphore 只能表达”有/无”，如果要传递数据，必须配合全局变量（volatile + 临界区）。Signal 把”通知”和”数据”合为一体。

10.3.5 Signal vs Channel（容量=1）#

// 这两个看起来很像：
static SIG: Signal = Signal::new();
static CH: Channel = Channel::new();
rust
行为SignalChannel<..., 1>
发送时已有值覆盖旧值阻塞等待空间
语义”最新状态是什么""每个事件都要处理”
丢失数据✅ 允许（只保留最新）❌ 不允许（FIFO 保证）
适用场景按键状态、模式切换传感器数据流、命令队列

选择规则：

  • 如果你关心”每一个”事件 → Channel
  • 如果你只关心”最新的”状态 → Signal

10.4 Mutex：互斥访问共享资源#

10.4.1 为什么 Embassy 还需要 Mutex？#

你可能会问：Embassy 是单线程协作式调度，为什么还需要 Mutex？

答案：因为有些资源被多个任务使用，而任务在 .await 点让出 CPU。

// ❌ 错误：两个任务共享同一个 I2C 实例
static I2C: I2c<'static> = /* ... */;

#[embassy_executor::task]
async fn task_a() {
    I2C.write_read(0x76, &[0xFA], &mut buf).await;  // ← await 点！
    // 在 await 期间，task_b 可能也在使用 I2C！
}

#[embassy_executor::task]
async fn task_b() {
    I2C.write_read(0x68, &[0x75], &mut buf).await;  // ← 冲突！
}
rust

问题：i2c.write_read().await 在 DMA 传输期间让出 CPU。如果此时另一个任务也发起 I2C 传输，两个 DMA 请求会冲突。

解决：用 Mutex 保护共享资源。

10.4.2 Embassy Mutex 的两种形式#

形式一：embassy_sync::mutex::Mutex（异步 Mutex）#

形式二：embassy_sync::blocking_mutex::Mutex（同步 Mutex）#

use embassy_sync::blocking_mutex::Mutex as BlockingMutex;
use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex;
use core::cell::RefCell;

// 用于保护不需要 .await 的简单数据
static COUNTER: BlockingMutex> =
    BlockingMutex::new(RefCell::new(0));

// 使用（同步，不挂起）
fn increment() {
    COUNTER.lock(|c| {
        *c.borrow_mut() += 1;
    });
    // 锁在闭包结束时自动释放
}
rust

10.4.3 两种 Mutex 的选择#

异步 Mutex同步 BlockingMutex
锁内可以 .await？✅ 可以❌ 不可以
获取锁时可能挂起？✅ 可能❌ 不会（临界区内）
适用场景保护 I2C/SPI/UART 等外设保护计数器、标志、配置
性能略低（需要挂起/恢复）极高（只是关/开中断）
死锁风险理论上存在（但协作式下极难触发）无（临界区不可嵌套等待）

10.4.4 完整示例：共享 UART 日志#

10.4.5 与 FreeRTOS Mutex 的对比#

方面FreeRTOS MutexEmbassy Mutex
创建xSemaphoreCreateMutex()static M: Mutex<..., T> = Mutex::new(...)
获取xSemaphoreTake(m, timeout)let guard = M.lock().await
释放xSemaphoreGive(m)（手动！）自动（guard drop）
忘记释放死锁（运行时）不可能（RAII，编译器保证）
优先级继承内置（防止优先级反转）不需要（无优先级）
递归锁需要 xSemaphoreCreateRecursiveMutex()不支持（也不需要）
锁内阻塞可以（但危险）异步 Mutex 可以 .await
类型安全❌（Mutex 和数据分离）✅（Mutex 包裹数据）

最关键的差异：RAII 自动释放

// C：忘记释放 Mutex → 死锁
void task_a(void) {
    xSemaphoreTake(mutex, portMAX_DELAY);
    do_something();
    if (error) {
        return;  // ❌ 忘记 xSemaphoreGive！其他任务永久阻塞！
    }
    xSemaphoreGive(mutex);
}
c
// Rust：不可能忘记释放
async fn task_a() {
    let guard = M.lock().await;
    do_something();
    if error {
        return;  // ✅ guard 自动 drop → 锁自动释放！
    }
    // guard 在这里 drop → 锁自动释放
}
rust

这就是 RAII（Resource Acquisition Is Initialization）的力量——资源的生命周期绑定到变量的生命周期。编译器保证：只要变量活着，资源就被持有；变量死了，资源就被释放。没有例外，没有”忘记”。

10.4.6 替代方案：所有权转移（无锁）#

在很多情况下，你可以完全避免 Mutex——通过所有权转移：

设计原则：

能用所有权解决的，不用 Mutex。必须共享的，才用 Mutex。

场景推荐方案
每个任务用不同的外设所有权转移（spawn 时传入）
多个任务用同一个外设Mutex 保护
多个任务用同一个外设的不同功能考虑 HAL 是否提供分离的句柄
简单的共享计数器/标志BlockingMutex<RefCell<T>>

10.5 PubSub：一对多广播#

10.5.1 什么是 PubSub？#

PubSub（Publish-Subscribe）是一个一对多广播通道：

  • 发布者：publish(value)——所有订阅者都会收到
  • 订阅者：next_message().await——等待新消息
  • 每个订阅者有独立的消息队列
  • 新订阅者不会收到订阅之前的消息
                                    ┌──────────────┐
                              ┌──►  │ Subscriber 1 │  (独立队列)
                              │     └──────────────┘
┌───────────┐   publish(v)   │     ┌──────────────┐
│ Publisher │ ───────────────┼──►  │ Subscriber 2 │  (独立队列)
└───────────┘                 │     └──────────────┘
                              │     ┌──────────────┐
                              └──►  │ Subscriber 3 │  (独立队列)
                                    └──────────────┘
plaintext

10.5.2 API 详解#

发布者#

// 获取发布者（最多 1 个，由泛型参数限制）
let publisher = EVENT_BUS.publisher().unwrap();

// 发布消息（永不阻塞，永不失败）
publisher.publish(SystemEvent::ButtonPressed);
publisher.publish(SystemEvent::SensorUpdated { temp: 253 });
rust

订阅者#

// 获取订阅者（最多 4 个）
let mut sub1 = EVENT_BUS.subscriber().unwrap();
let mut sub2 = EVENT_BUS.subscriber().unwrap();

// 异步等待消息
let event = sub1.next_message().await;  // 无新消息时挂起

// 非阻塞检查
match sub1.try_next_message() {
    Some(event) => { /* 有新消息 */ }
    None => { /* 无新消息 */ }
}
rust

10.5.3 完整示例：系统事件总线#

10.5.4 与 FreeRTOS Event Group + Queue 的对比#

在 FreeRTOS 中实现一对多广播，通常需要组合使用 Event Group 和 Queue：

FreeRTOS 方案的问题：

  1. 多个订阅者竞争同一个 Queue → 只有一个能收到消息
  2. 需要为每个订阅者创建独立的 Queue → 发布者要向所有 Queue 发送
  3. Event Group 只有 24 位 → 事件类型有限
  4. 组合使用复杂，容易出错
方面FreeRTOS（Event Group + Queue）Embassy PubSub
一对多需要手动实现（每订阅者一个 Queue）内置（自动分发）
订阅者独立性需要手动管理自动（每订阅者独立队列）
事件携带数据需要额外 Queue内置（泛型类型）
新订阅者需要修改发布者代码无需（动态订阅）
代码量~50 行~10 行

10.5.5 PubSub vs 多个 Channel#

// 方案 A：PubSub（推荐用于广播）
static BUS: PubSubChannel<..., Event, 4, 4, 1> = PubSubChannel::new();
// 发布者只需一行：
publisher.publish(event);

// 方案 B：多个 Channel（手动广播）
static CH1: Channel<..., Event, 4> = Channel::new();
static CH2: Channel<..., Event, 4> = Channel::new();
static CH3: Channel<..., Event, 4> = Channel::new();
// 发布者需要向每个 Channel 发送：
CH1.send(event).await;
CH2.send(event).await;
CH3.send(event).await;
// 新增订阅者？修改发布者代码！
rust
PubSub多个 Channel
新增订阅者无需修改发布者需要修改发布者
RAM 开销共享发布逻辑每个 Channel 独立
语义清晰度”广播""点对点 × N”
适用场景事件通知、状态变更不同数据流

10.6 选型指南：如何选择正确的原语#

10.6.1 决策树#

10.6.2 完整对比表#

原语容量方向阻塞语义数据典型场景
Channel<T, N>N（编译期）多对多满→发送方挂起；空→接收方挂起✅ 类型化传感器数据流、命令队列
Signal<T>1（覆盖）多对一空→接收方挂起；满→覆盖✅ 类型化按键事件、模式切换
Mutex<T>1（资源）互斥被占用→请求方挂起✅ 包裹资源共享 I2C/SPI/UART
PubSub<T, N, S>N/订阅者一对多空→订阅者挂起；满→覆盖最旧✅ 类型化系统事件总线

10.6.3 常见模式速查#

模式原语示例
生产者-消费者Channel传感器 → 处理器
最新值缓存Signal按键状态、当前模式
事件广播PubSub系统事件 → LED/日志/显示
资源共享Mutex多任务共用 UART
请求-响应Channel × 2命令 Channel + 响应 Channel
流水线Channel 链采集 → 滤波 → 显示
中断通知任务SignalISR → 任务（按键、DMA 完成）

10.6.4 反模式：不要这样做#

10.7 综合实战：多任务数据采集系统#

需求#

  • 2 个传感器（I2C）以不同速率采集
  • 数据经过滤波后存入环形缓冲区
  • OLED 显示最新数据
  • UART 接收命令（“START”/“STOP”/“GET”）
  • 按键切换显示页面
  • 所有任务通过事件总线协调

架构图#

完整代码#

通信原语使用总结#

原语实例连接作用
ChannelSENSOR_CH传感器 → 滤波原始数据流
ChannelFILTERED_CH滤波 → 显示处理后数据
PubSubEVENT_BUS按键/传感器 → 显示/日志事件广播
SignalCMD_SIGNALUART → 控制逻辑命令通知
MutexRUNNINGUART ↔ 传感器共享运行状态

10.8 本章小结#

核心认知#

  1. Channel 是工作马：80% 的任务间通信需求可以用 Channel 解决。它是类型安全的、固定容量的、异步的 FIFO 队列。

  2. Signal 是”最新值”语义：当你只关心”当前状态是什么”而非”发生了什么事件”时，用 Signal。它永远不会阻塞发送方。

  3. Mutex 是最后手段：优先考虑所有权转移（每个任务拥有自己的外设）。只有真正需要共享时，才用 Mutex。RAII 保证你不会忘记释放。

  4. PubSub 是广播：当一个事件需要通知多个不相关的任务时，PubSub 比”多个 Channel”更优雅、更可扩展。

  5. 没有死锁：Embassy 的协作式调度 + 单线程 Executor 意味着死锁在理论上不可能发生（除非你在 Mutex 锁内 .await 同一个 Mutex）。

与 FreeRTOS 的最终对比#

维度FreeRTOSEmbassy
类型安全❌ void*✅ 泛型
内存分配动态（pvPortMalloc）静态（编译期）
资源释放手动（xSemaphoreGive）自动（RAII drop）
死锁可能（运行时）不可能（编译期+协作式）
优先级反转可能（需优先级继承）不存在（无优先级）
竞态条件可能（volatile 不够）编译器阻止（借用检查）
代码量多（初始化+发送+接收+错误处理）少（声明+使用）

下一步#

你现在掌握了 Embassy 的完整编程模型：

  • ✅ 外设操作（第 8 章）
  • ✅ 运行时机制（第 9 章）
  • ✅ 任务间通信（第 10 章）

接下来，我们将把这些知识组合成一个完整的、可部署的项目：

第 11 章：完整项目实战——从传感器到云端。我们将构建一个包含传感器采集、OLED 显示、UART 调试、WiFi 上报、OTA 升级的完整 IoT 产品原型。

下一章：第 11 章——完整项目实战：从传感器到云端。

上一章 第 9 章：Embassy 运行时——Executor、Future 与 Waker