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

本章要回答的问题：Embassy 的任务之间怎么安全通信？Signal、Channel、Mutex 各自适合什么场景？如何避免死锁和饥饿？

10.1 为什么需要任务间通信？#

任务隔离：Embassy 的设计哲学#

在第 9 章中，我们了解了 Embassy 的任务模型：每个任务是一个独立的 async fn，拥有自己的局部状态。任务之间不共享栈空间，也不像 RTOS 那样通过全局变量随意交换数据。这种隔离是有意为之的——它让每个任务的逻辑自包含、可推理、可测试。

但现实世界的应用不可能完全隔离。考虑一个典型的 IoT 传感器节点：

┌──────────────┐     ┌──────────────┐     ┌──────────────┐
│  传感器采集   │────▶│  数据处理     │────▶│  无线发送     │
│  Task        │     │  Task        │     │  Task        │
└──────────────┘     └──────────────┘     └──────────────┘
     100ms                滤波/校准             BLE/WiFi

plaintext

三个任务各司其职，但数据必须从采集任务流向处理任务，再流向发送任务。这就是任务间通信（Inter-Task Communication）要解决的问题。

C 工程师的”老办法”及其问题#

在 C 的裸机或 RTOS 开发中，任务间通信通常依赖以下手段：

C 的做法	问题
全局变量 + `volatile`	无类型安全，无访问控制，多写者竞态
中断回调函数指针	回调中不能阻塞，逻辑碎片化
环形缓冲区 + 标志位	手动管理读写指针，容易出 bug
FreeRTOS Queue	功能完整，但 API 繁琐，类型不安全（`void*`）
FreeRTOS Task Notification	轻量但仅限一对一，语义有限

这些方案的共同问题是：正确性依赖程序员的纪律，而非编译器的保证。一个 volatile 变量被两个任务同时写入，编译器不会报错；一个队列的 void* 被错误地强转，运行时才会崩溃。

Embassy 的答案：类型安全的通信原语#

embassy-sync 库提供了一组专为嵌入式异步环境设计的通信原语。它们的核心特点是：

类型安全：通道中传递的数据有明确类型，编译期检查
零堆分配：所有原语使用静态存储，适合 no_std 环境
异步友好：等待操作是 .await，不阻塞 CPU
编译期约束：通过 Rust 类型系统阻止非法使用（如跨 Executor 共享）

embassy-sync 的模块全景：

embassy-sync
├── signal          // 单值信号（最新值覆盖）
├── watch           // 多消费者观察通道
├── channel         // MPMC 有界队列
├── priority_channel// 带优先级的队列
├── pubsub          // 发布-订阅通道
├── mutex           // 异步互斥锁
├── pipe            // 异步字节管道
├── blocking_mutex  // 底层互斥原语（临界区/线程模式）
└── lazy_lock       // 延迟初始化

plaintext

在深入每个原语之前，我们需要理解一个贯穿所有 API 的泛型参数：M（Raw Mutex）。

理解 `M`：互斥锁类型参数#

你会注意到 embassy-sync 中几乎所有类型都带有一个泛型参数 M：

Signal<M, T>
Channel<M, T, N>
Mutex<M, T>

rust

这个 M 决定了该原语在哪些执行上下文之间共享。embassy-sync 提供了两种主要的 Raw Mutex 实现：

类型	含义	适用场景
`CriticalSectionRawMutex`	使用临界区（关中断）保护内部状态	需要在任务与中断之间共享
`ThreadModeRawMutex`	仅在 Thread Mode 中有效，不关中断	仅在同一 Executor 的任务之间共享，性能更高

C 工程师的类比：CriticalSectionRawMutex 相当于 __disable_irq() / __enable_irq() 包裹的临界区；ThreadModeRawMutex 相当于”约定只在主循环中访问，不需要保护”。

use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex;
use embassy_sync::blocking_mutex::raw::ThreadModeRawMutex;

// 可在中断和任务之间共享
type MySignal = Signal<CriticalSectionRawMutex, u32>;

// 仅在同一 Executor 的任务之间共享（更高效）
type MyChannel = Channel<ThreadModeRawMutex, SensorData, 8>;

rust

实践建议：如果你不确定该用哪个，先用 CriticalSectionRawMutex。它的开销极小（几条指令的关/开中断），但保证了所有场景下的正确性。只有在性能分析确认临界区是瓶颈时，才考虑切换到 ThreadModeRawMutex。

10.2 `Signal`——单生产者单消费者信号#

语义：最新值覆盖旧值#

Signal 是 embassy-sync 中最简单的通信原语。它在任意时刻只保存一个值。当生产者发送新值时，旧值被覆盖。消费者调用 .wait() 时，获取当前最新的值。

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

// 定义一个传递 u32 的信号
static SENSOR_READY: Signal<CriticalSectionRawMutex, u32> = Signal::new();

rust

与 C 的 `volatile` 标志位对比#

C 工程师最熟悉的做法：

// C 的做法
volatile uint32_t sensor_value = 0;
volatile uint8_t  sensor_ready = 0;

// 中断中写入
void ADC_IRQHandler(void) {
    sensor_value = ADC1->DR;
    sensor_ready = 1;
}

// 主循环中轮询
while (1) {
    if (sensor_ready) {
        sensor_ready = 0;
        process(sensor_value);
    }
    // 问题：如果两次中断之间主循环没来得及检查，
    // 第一次的值就丢失了（被覆盖）
}

Embassy 的 Signal 本质上做的是同样的事——最新值覆盖——但有两个关键改进：

不需要轮询：消费者 .await 挂起，有新值时自动唤醒
类型安全：不可能把 u32 误读为 float

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

static ADC_VALUE: Signal<CriticalSectionRawMutex, u32> = Signal::new();

// 生产者：中断或高优先级任务
#[embassy_executor::task]
async fn adc_task(mut adc: Adc<'static, ADC1>) {
    loop {
        let value = adc.read(&mut pin).await;
        ADC_VALUE.signal(value);  // 发送信号，覆盖旧值
        Timer::after_millis(100).await;
    }
}

// 消费者：处理任务
#[embassy_executor::task]
async fn process_task() {
    loop {
        let value = ADC_VALUE.wait().await;  // 挂起等待，有新值时唤醒
        defmt::info!("ADC value: {}", value);
        // 处理数据...
    }
}

rust

API 速览#

方法	说明	是否异步
`signal(value)`	发送值，覆盖旧值，唤醒等待者	否
`wait()`	等待并获取值（消费后清空）	是（`.await`）
`try_take()`	非阻塞尝试获取，返回 `Option<T>`	否
`signaled()`	检查是否有未消费的值	否

适用场景#

状态通知：某个事件发生了（不关心中间过程，只关心最新状态）
最新值传递：传感器数据、电池电压等”只关心当前值”的场景
中断到任务的通知：中断中调用 signal()，任务中 .await 等待

不适用场景#

需要接收每一个值（中间值不能丢失）→ 用 Channel
多个消费者需要各自独立接收 → 用 PubSubChannel 或 Watch

10.3 `Watch`——多消费者观察通道#

语义：多个读者观察同一个值的变化#

Watch 与 Signal 类似，都保存”最新值”。关键区别在于：

特性	`Signal`	`Watch`
消费者数量	一个（值被消费后清空）	多个（每个消费者独立跟踪）
值的生命周期	消费后消失	始终保留最新值
消费者行为	`wait()` 获取值后，信号清空	`get()` 获取当前值，不清空

与 C 的”多个中断读取同一个全局变量”对比#

// C 的做法：多个模块读取同一个配置
volatile system_config_t g_config;

// 模块 A：显示任务
void display_task(void) {
    while (1) {
        system_config_t cfg = g_config;  // 读取
        update_display(cfg.brightness, cfg.contrast);
        vTaskDelay(100);
    }
}

// 模块 B：背光控制
void backlight_task(void) {
    while (1) {
        system_config_t cfg = g_config;  // 读取
        set_pwm(cfg.brightness);
        vTaskDelay(50);
    }
}

// 问题：如何知道配置"变了"？只能轮询或额外加标志位

Watch 提供了优雅的解决方案：每个消费者可以 .await 等待值变化，变化时自动唤醒。

use embassy_sync::watch::Watch;
use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex;

#[derive(Clone, Copy, PartialEq)]
struct SystemConfig {
    brightness: u8,
    volume: u8,
}

static CONFIG: Watch<CriticalSectionRawMutex, SystemConfig> = Watch::new();

// 生产者：配置更新任务（例如来自 BLE 命令）
#[embassy_executor::task]
async fn config_updater() {
    loop {
        let new_config = receive_ble_config().await;
        CONFIG.send(new_config);  // 更新值，唤醒所有等待者
    }
}

// 消费者 A：显示任务
#[embassy_executor::task]
async fn display_task() {
    let mut reader = CONFIG.reader();  // 获取独立的读取器
    loop {
        let config = reader.get().await;  // 等待值变化
        update_display(config.brightness);
    }
}

// 消费者 B：背光控制
#[embassy_executor::task]
async fn backlight_task() {
    let mut reader = CONFIG.reader();  // 另一个独立的读取器
    loop {
        let config = reader.get().await;  // 独立跟踪变化
        set_pwm(config.brightness);
    }
}

rust

关键概念：Reader#

每个消费者通过 CONFIG.reader() 获取一个独立的 Reader。每个 Reader 内部维护自己的”已读版本号”，因此：

消费者 A 读取后，不影响消费者 B
如果值在消费者 A 两次 get() 之间变化了多次，A 只会看到最新值（中间值跳过）
如果值没有变化，get().await 会挂起

适用场景#

配置广播：系统配置变更通知多个子系统
传感器数据广播：一个采集任务，多个处理任务各自消费
状态同步：设备状态（在线/离线/错误）通知多个 UI 组件

10.4 `Channel`——MPMC 有界队列#

语义：生产者-消费者队列#

Channel 是 embassy-sync 中最常用的通信原语，语义上等价于 FreeRTOS 的 Queue。它是一个有界的、FIFO 的、多生产者多消费者（MPMC）队列。

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

// 定义：传递 SensorReading 类型，容量为 8 的通道
static SENSOR_CHAN: Channel<CriticalSectionRawMutex, SensorReading, 8> = Channel::new();

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

rust

与 FreeRTOS Queue 的对比#

// FreeRTOS 的做法
QueueHandle_t xSensorQueue;

void vSensorTask(void *pvParameters) {
    SensorReading_t reading;
    xSensorQueue = xQueueCreate(8, sizeof(SensorReading_t));
    
    while (1) {
        read_sensor(&reading);
        // 发送：如果队列满，阻塞等待最多 100ms
        xQueueSend(xSensorQueue, &reading, pdMS_TO_TICKS(100));
    }
}

void vProcessTask(void *pvParameters) {
    SensorReading_t reading;
    while (1) {
        // 接收：如果队列空，永久阻塞
        if (xQueueReceive(xSensorQueue, &reading, portMAX_DELAY) == pdPASS) {
            process(reading);
        }
    }
}

Embassy 的等价实现：

// 生产者任务
#[embassy_executor::task]
async fn sensor_task() {
    let mut sensor = init_sensor();
    loop {
        let reading = sensor.read().await;
        // 发送：如果队列满，.await 挂起（不阻塞 CPU）
        SENSOR_CHAN.send(reading).await;
    }
}

// 消费者任务
#[embassy_executor::task]
async fn process_task() {
    loop {
        // 接收：如果队列空，.await 挂起
        let reading = SENSOR_CHAN.receive().await;
        process(reading);
    }
}

rust

关键差异#

特性	FreeRTOS Queue	Embassy Channel
类型安全	`void*`，运行时可能类型错误	泛型 `T`，编译期保证
内存分配	`xQueueCreate` 动态分配	静态分配，编译期确定
阻塞行为	阻塞整个线程（RTOS 调度器切换）	挂起当前 Future（零开销）
中断安全	需要 `FromISR` 后缀 API	`CriticalSectionRawMutex` 自动处理
容量	运行时指定	编译期常量（泛型参数 `N`）

API 速览#

方法	说明	是否异步
`send(value).await`	发送值，队列满时挂起	是
`receive().await`	接收值，队列空时挂起	是
`try_send(value)`	非阻塞发送，返回 `Result<(), TrySendError<T>>`	否
`try_receive()`	非阻塞接收，返回 `Result<T, TryReceiveError>`	否
`len()`	当前队列中的元素数量	否
`is_empty()` / `is_full()`	状态查询	否

中断中使用 Channel#

在 C 中，中断里调用 xQueueSend 是危险的（必须用 xQueueSendFromISR）。在 Embassy 中，使用 CriticalSectionRawMutex 的 Channel 可以在中断中安全调用 try_send：

#[interrupt]
fn EXTI0() {
    // 中断中不能用 .await，用 try_send
    let _ = BUTTON_CHAN.try_send(ButtonEvent::Pressed);
}

rust

注意：中断中只能使用 try_send / try_receive（非阻塞版本），不能使用 .await。

适用场景#

数据流处理：传感器 → 滤波 → 上报
命令队列：UI 输入 → 命令解析 → 执行
日志收集：多个任务产生日志 → 统一输出任务

10.5 `PriorityChannel`——优先级通道#

语义：高优先级消息插队#

PriorityChannel 是 Channel 的增强版本。当队列中有多个消息等待消费时，优先级高的消息先被取出。

use embassy_sync::priority_channel::{PriorityChannel, Priority};
use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex;

#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
enum Command {
    EmergencyStop,    // 最高优先级
    FaultReport,      // 高优先级
    NormalReading,    // 普通优先级
    DebugInfo,        // 最低优先级
}

// 实现 Priority trait
impl Priority for Command {
    fn priority(&self) -> u8 {
        match self {
            Command::EmergencyStop => 3,
            Command::FaultReport => 2,
            Command::NormalReading => 1,
            Command::DebugInfo => 0,
        }
    }
}

static CMD_CHAN: PriorityChannel<CriticalSectionRawMutex, Command, 16> = PriorityChannel::new();

rust

工作原理#

PriorityChannel 内部维护多个子队列（按优先级分层）。receive().await 总是从最高优先级的非空子队列中取出消息。

┌─────────────────────────────────────────┐
│           PriorityChannel               │
│                                         │
│  Priority 3: [EmergencyStop]            │  ← 先消费
│  Priority 2: [FaultReport, FaultReport] │  ← 其次
│  Priority 1: [NormalReading, ...]       │  ← 再次
│  Priority 0: [DebugInfo, DebugInfo, ...]│  ← 最后
└─────────────────────────────────────────┘

plaintext

适用场景#

紧急命令：急停、故障报警必须优先处理
混合数据流：控制命令（高优先级）与遥测数据（低优先级）共用一个通道
QoS 调度：不同优先级的网络包处理

注意事项#

低优先级消息可能饥饿：如果高优先级消息持续到来，低优先级消息永远得不到处理
解决方案：设置容量限制，或在消费者中实现”加权轮询”策略

10.6 `PubSubChannel`——发布-订阅通道#

语义：一对多广播，每个订阅者独立消费#

PubSubChannel 实现了经典的发布-订阅（Publish-Subscribe）模式。发布者发送的消息会被所有订阅者各自独立接收。

use embassy_sync::pubsub::{PubSubChannel, Subscriber};
use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex;

// 参数：Mutex类型, 消息类型, 通道容量, 最大订阅者数, 最大发布者数
static EVENT_BUS: PubSubChannel<CriticalSectionRawMutex, SystemEvent, 16, 4, 1> = PubSubChannel::new();

#[derive(Clone, Copy)]
enum SystemEvent {
    ButtonPressed(u8),
    UsbConnected,
    UsbDisconnected,
    BatteryLow(u8),  // 剩余百分比
}

rust

与 C 的回调函数列表对比#

// C 的做法：手动维护回调列表
typedef void (*event_callback_t)(SystemEvent_t event);

#define MAX_SUBSCRIBERS 4
static event_callback_t subscribers[MAX_SUBSCRIBERS];
static uint8_t subscriber_count = 0;

void register_callback(event_callback_t cb) {
    if (subscriber_count < MAX_SUBSCRIBERS) {
        subscribers[subscriber_count++] = cb;
    }
}

void notify_all(SystemEvent_t event) {
    for (int i = 0; i < subscriber_count; i++) {
        subscribers[i](event);  // 问题：回调中不能阻塞！
    }
}

C 的回调方式有严重限制：回调函数中不能阻塞、不能做耗时操作。PubSubChannel 彻底解决了这个问题——每个订阅者是一个独立的异步任务，可以按自己的节奏消费消息。

// 发布者
#[embassy_executor::task]
async fn event_producer() {
    loop {
        let event = detect_event().await;
        // 发布消息给所有订阅者
        EVENT_BUS.publish(event).await;
    }
}

// 订阅者 A：LED 指示
#[embassy_executor::task]
async fn led_indicator() {
    let mut sub = EVENT_BUS.subscriber().unwrap();
    loop {
        let event = sub.next_message().await;
        match event {
            SystemEvent::UsbConnected => led.set_high(),
            SystemEvent::UsbDisconnected => led.set_low(),
            _ => {}
        }
    }
}

// 订阅者 B：日志记录
#[embassy_executor::task]
async fn event_logger() {
    let mut sub = EVENT_BUS.subscriber().unwrap();
    loop {
        let event = sub.next_message().await;
        defmt::info!("Event: {:?}", event);
    }
}

// 订阅者 C：BLE 通知
#[embassy_executor::task]
async fn ble_notifier() {
    let mut sub = EVENT_BUS.subscriber().unwrap();
    loop {
        let event = sub.next_message().await;
        ble_notify(event).await;  // 可以安全地做耗时操作
    }
}

rust

关键行为：慢订阅者#

如果某个订阅者消费速度跟不上发布速度，会发生什么？

当通道满时，最旧的消息会被丢弃（对于慢订阅者而言）
订阅者会收到一个 Lagged(n) 通知，表示它错过了 n 条消息
其他订阅者不受影响

loop {
    match sub.next_message().await {
        Message::Payload(event) => process(event),
        Message::Lagged(n) => {
            defmt::warn!("Missed {} events!", n);
        }
    }
}

rust

适用场景#

事件总线：系统事件广播给多个处理模块
日志分发：一条日志同时输出到串口、BLE、SD 卡
状态变更通知：设备状态变化通知 UI、通信、存储等子系统

10.7 `Mutex`——异步互斥锁#

与 C 的互斥锁对比#

// FreeRTOS 的做法
SemaphoreHandle_t xMutex;
SharedData_t shared_data;

void vTaskA(void *pvParameters) {
    while (1) {
        xSemaphoreTake(xMutex, portMAX_DELAY);  // 获取锁
        shared_data.counter++;                   // 访问共享数据
        shared_data.buffer[shared_data.index] = read_sensor();
        xSemaphoreGive(xMutex);                  // 释放锁
        vTaskDelay(10);
    }
}

Embassy 的异步 Mutex：

use embassy_sync::mutex::Mutex;
use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex;

struct SharedState {
    counter: u32,
    buffer: [u16; 64],
    index: usize,
}

static SHARED: Mutex<CriticalSectionRawMutex, SharedState> = Mutex::new(SharedState {
    counter: 0,
    buffer: [0; 64],
    index: 0,
});

#[embassy_executor::task]
async fn task_a() {
    loop {
        // 获取锁：如果锁被占用，.await 挂起
        let mut state = SHARED.lock().await;
        state.counter += 1;
        state.buffer[state.index] = read_sensor().await;  // ⚠️ 危险！见下文
        state.index = (state.index + 1) % 64;
        // state 离开作用域时自动释放锁（RAII）
        
        Timer::after_millis(10).await;
    }
}

rust

⚠️ 异步 Mutex 的黄金规则：持有锁时不要 `.await`#

这是 C 工程师最容易犯的错误。在上面的代码中，read_sensor().await 在持有锁的情况下挂起了当前任务。这意味着：

当前任务挂起，但锁没有被释放
其他任务尝试 SHARED.lock().await 时会一直等待
如果 read_sensor() 需要另一个任务先完成某操作 → 死锁

正确做法：

#[embassy_executor::task]
async fn task_a() {
    loop {
        // 先在锁外完成异步操作
        let sensor_value = read_sensor().await;
        
        // 再获取锁，只做纯同步的数据操作
        let mut state = SHARED.lock().await;
        state.counter += 1;
        state.buffer[state.index] = sensor_value;
        state.index = (state.index + 1) % 64;
        // 锁在这里释放
        
        Timer::after_millis(10).await;
    }
}

rust

为什么 Embassy 的 Mutex 不像 tokio 那样”检测”死锁？#

在 std 环境（如 tokio）中，有些 Mutex 实现会在检测到死锁时 panic。但嵌入式环境中：

没有运行时检测的余裕（RAM 和 CPU 都有限）
panic 通常意味着系统重启
正确性应该由设计保证，而非运行时检测

`RefCell` 替代方案：单 Executor 场景#

如果共享数据只在同一个 Executor 的任务之间访问（不涉及中断），可以使用更轻量的方案：

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

// ThreadModeRawMutex + RefCell：零开销的内部可变性
static COUNTER: BlockingMutex<ThreadModeRawMutex, RefCell<u32>> = 
    BlockingMutex::new(RefCell::new(0));

#[embassy_executor::task]
async fn task_a() {
    loop {
        COUNTER.lock(|c| {
            *c.borrow_mut() += 1;
        });
        Timer::after_millis(100).await;
    }
}

rust

这种方式没有 .await，锁的获取和释放是同步的（临界区极短），适合保护简单的共享状态。

适用场景#

共享外设：多个任务需要访问同一个 SPI 设备
共享状态：系统配置、运行统计等
资源池：有限的 DMA 通道、缓冲区等

10.8 `Pipe`——异步字节管道#

语义：面向字节流的异步管道#

Pipe 是 embassy-sync 中专门用于字节流通信的原语。它的语义类似于 Unix 的管道或 C 的环形缓冲区，但完全异步。

use embassy_sync::pipe::Pipe;
use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex;

// 容量为 256 字节的管道
static DATA_PIPE: Pipe<CriticalSectionRawMutex, 256> = Pipe::new();

rust

与 C 的环形缓冲区对比#

// C 的做法：手动管理环形缓冲区
typedef struct {
    uint8_t buffer[256];
    volatile uint16_t head;
    volatile uint16_t tail;
} RingBuffer_t;

// 问题：
// 1. 读写指针的原子性需要手动保证
// 2. 满/空判断容易出错（head == tail 是空还是满？）
// 3. 阻塞等待需要手动实现（信号量 + 标志位）

Embassy 的 Pipe 将这些复杂性全部封装：

// 生产者：从 UART 接收数据写入管道
#[embassy_executor::task]
async fn uart_receiver(mut uart: Uart<'static, UART1>) {
    let mut buf = [0u8; 64];
    loop {
        let n = uart.read(&mut buf).await.unwrap();
        // 写入管道：如果管道满，.await 挂起
        DATA_PIPE.write_all(&buf[..n]).await;
    }
}

// 消费者：从管道读取数据，解析协议
#[embassy_executor::task]
async fn protocol_parser() {
    let mut header = [0u8; 4];
    loop {
        // 读取固定长度的头部
        DATA_PIPE.read_exact(&mut header).await;
        let len = u16::from_le_bytes([header[2], header[3]]) as usize;
        
        // 读取变长载荷
        let mut payload = [0u8; 128];
        DATA_PIPE.read_exact(&mut payload[..len]).await;
        
        process_packet(&header, &payload[..len]);
    }
}

rust

API 速览#

方法	说明
`write(buf).await`	写入字节，返回实际写入长度
`write_all(buf).await`	写入全部字节，空间不足时挂起
`read(buf).await`	读取字节，返回实际读取长度
`read_exact(buf).await`	读满缓冲区，数据不足时挂起
`try_write(buf)`	非阻塞写入
`try_read(buf)`	非阻塞读取

适用场景#

协议解析：UART/SPI 原始字节流 → 帧解析
数据缓冲：DMA 接收 → 处理任务
日志缓冲：多个任务写入 → 统一输出

10.9 与 FreeRTOS 通信原语的完整对比表#

Embassy 原语	FreeRTOS 对应	语义差异	优势
`Signal`	Task Notification (`eSetValueWithOverwrite`)	几乎相同	类型安全，无需手动管理通知值
`Watch`	无直接对应（需全局变量 + Event Group）	Watch 自带变更通知	多消费者独立跟踪，无需轮询
`Channel`	Queue	几乎相同	类型安全，静态分配，无需 `FromISR` 变体
`PriorityChannel`	无直接对应（需多个 Queue + 手动调度）	内置优先级	一个原语解决，无需手动管理
`PubSubChannel`	Event Group（部分对应）	Event Group 只有标志位，无数据	携带数据，每个订阅者独立消费
`Mutex`	Mutex Semaphore	几乎相同	RAII 自动释放，不会忘记 give
`Pipe`	Stream Buffer	几乎相同	类型安全（字节流），API 更简洁
`blocking_mutex::Mutex`	临界区（`taskENTER_CRITICAL`）	同步锁，不挂起	零开销，适合极短临界区

一个完整的对比示例#

需求：按键中断触发，通知主任务处理。

FreeRTOS 版本：

// 全局变量
TaskHandle_t xMainTaskHandle;

// 中断
void EXTI0_IRQHandler(void) {
    BaseType_t xHigherPriorityTaskWoken = pdFALSE;
    vTaskNotifyGiveFromISR(xMainTaskHandle, &xHigherPriorityTaskWoken);
    portYIELD_FROM_ISR(xHigherPriorityTaskWoken);
}

// 任务
void vMainTask(void *pvParameters) {
    while (1) {
        ulTaskNotifyTake(pdTRUE, portMAX_DELAY);
        handle_button();
    }
}

Embassy 版本：

static BUTTON_SIGNAL: Signal<CriticalSectionRawMutex, ()> = Signal::new();

#[interrupt]
fn EXTI0() {
    BUTTON_SIGNAL.signal(());
}

#[embassy_executor::task]
async fn main_task() {
    loop {
        BUTTON_SIGNAL.wait().await;
        handle_button().await;
    }
}

rust

对比要点：

不需要手动管理 TaskHandle
不需要 xHigherPriorityTaskWoken 和 portYIELD_FROM_ISR
不需要区分”任务级 API”和”中断级 API”
类型系统保证 () 不会被误用为其他类型

10.10 常见死锁与饥饿问题#

死锁模式 1：持有 Mutex 时 `.await`#

// ❌ 错误：持有锁时挂起
async fn bad_example() {
    let mut state = SHARED.lock().await;
    some_async_operation().await;  // 挂起，但锁未释放！
    state.value = 42;
}

// ✅ 正确：先完成异步操作，再获取锁
async fn good_example() {
    let result = some_async_operation().await;
    let mut state = SHARED.lock().await;
    state.value = result;
}

rust

死锁模式 2：嵌套锁#

// ❌ 错误：两个任务以不同顺序获取两把锁
// 任务 A
let mut lock1 = MUTEX_1.lock().await;
let mut lock2 = MUTEX_2.lock().await;  // 如果任务 B 已持有 MUTEX_2...

// 任务 B
let mut lock2 = MUTEX_2.lock().await;
let mut lock1 = MUTEX_1.lock().await;  // 死锁！

rust

解决方案：始终按固定顺序获取锁（如按地址从小到大），或重新设计避免嵌套。

死锁模式 3：Channel 循环依赖#

// ❌ 错误：两个 Channel 形成循环
// 任务 A：从 chan1 接收，处理后发送到 chan2
// 任务 B：从 chan2 接收，处理后发送到 chan1
// 如果两个 Channel 都满了 → 死锁

rust

解决方案：确保数据流是单向的（DAG），或为所有 send 设置超时。

饥饿问题#

场景	原因	解决方案
`PriorityChannel` 低优先级消息永远不被消费	高优先级消息持续到来	限制高优先级消息的产生速率
`PubSubChannel` 慢订阅者丢失消息	消费速度 < 发布速度	增大通道容量，或优化消费者
协作式调度中某任务不让出 CPU	任务中没有 `.await` 点	在循环中插入 `yield_now().await`

最佳实践清单#

持有 Mutex 的时间尽可能短——只保护必要的共享状态操作
永远不要在持有 Mutex 时 .await——除非你 100% 确定不会死锁
Channel 容量根据最坏情况设计——考虑生产者突发速率
优先使用消息传递（Channel/Signal）而非共享状态（Mutex）——这是 Rust 并发哲学的核心
为所有可能阻塞的操作设置超时——使用 embassy_time::with_timeout

use embassy_time::{with_timeout, Duration};

// 带超时的接收
match with_timeout(Duration::from_millis(1000), CHANNEL.receive()).await {
    Ok(value) => process(value),
    Err(_) => defmt::warn!("Receive timeout!"),
}

rust

10.11 选型指南#

决策表#

你的需求	推荐原语	理由
通知”某事发生了”，不关心次数	`Signal`	最简单，最新值覆盖
广播配置/状态给多个消费者	`Watch`	多读者独立跟踪变化
传递数据流，每条消息都要处理	`Channel`	FIFO 保证，不丢消息
数据流中有紧急消息需要插队	`PriorityChannel`	内置优先级调度
一个事件通知多个独立处理者	`PubSubChannel`	每个订阅者独立消费
多个任务共享一个可变状态	`Mutex`	互斥访问
字节流缓冲（UART/SPI 数据）	`Pipe`	面向字节流的异步读写
中断中通知任务	`Signal` + `CriticalSectionRawMutex`	中断中调用 `signal()`，任务中 `.await`
极短临界区保护简单变量	`blocking_mutex::Mutex` + `RefCell`	零异步开销

性能与资源开销对比#

原语	RAM 开销（典型）	发送延迟	接收延迟
`Signal<u32>`	~12 bytes	即时	即时（有值时）
`Watch<Config>`	~20 bytes + 每 Reader ~8 bytes	即时	即时（有变化时）
`Channel<T, 8>`	8 × sizeof(T) + ~16 bytes	即时（未满时）	即时（非空时）
`PriorityChannel<T, 16>`	16 × sizeof(T) + ~32 bytes	即时	即时
`PubSubChannel<T, 16, 4, 1>`	16 × sizeof(T) + ~48 bytes	即时	即时
`Mutex<T>`	sizeof(T) + ~8 bytes	N/A	即时（未锁时）
`Pipe<256>`	256 + ~16 bytes	即时（有空间时）	即时（有数据时）

注：以上数据为 Cortex-M4 上的典型值，实际取决于对齐和编译器优化。所有原语的”即时”操作都是 O(1) 时间复杂度。

组合使用模式#

实际项目中，通常需要组合使用多种原语：

// 典型 IoT 节点的通信架构

// 1. 按键中断 → 主任务（Signal）
static BUTTON: Signal<CriticalSectionRawMutex, ButtonEvent> = Signal::new();

// 2. 传感器 → 处理任务（Channel，保证每条数据都处理）
static SENSOR_DATA: Channel<CriticalSectionRawMutex, SensorReading, 16> = Channel::new();

// 3. 系统状态 → 多个 UI/通信任务（Watch）
static SYS_STATE: Watch<CriticalSectionRawMutex, SystemState> = Watch::new();

// 4. 事件广播（PubSubChannel）
static EVENTS: PubSubChannel<CriticalSectionRawMutex, Event, 8, 4, 2> = PubSubChannel::new();

// 5. 共享外设（Mutex）
static SPI_BUS: Mutex<CriticalSectionRawMutex, SpiDevice> = Mutex::new(/* ... */);

// 6. UART 原始数据缓冲（Pipe）
static UART_PIPE: Pipe<CriticalSectionRawMutex, 512> = Pipe::new();

rust

10.12 实战：完整的多任务传感器系统#

让我们把本章学到的所有原语组合起来，构建一个完整的多任务系统：

#![no_std]
#![no_main]

use embassy_executor::Spawner;
use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex;
use embassy_sync::channel::Channel;
use embassy_sync::mutex::Mutex;
use embassy_sync::pubsub::PubSubChannel;
use embassy_sync::signal::Signal;
use embassy_sync::watch::Watch;
use embassy_time::{Duration, Timer};
use defmt::*;

// ============ 共享状态定义 ============

#[derive(Clone, Copy)]
struct SensorReading {
    temperature: i16,  // 0.1°C 单位
    humidity: u16,     // 0.1% 单位
}

#[derive(Clone, Copy, PartialEq)]
enum SystemMode {
    Normal,
    LowPower,
    Alarm,
}

#[derive(Clone, Copy)]
enum SystemEvent {
    ModeChanged(SystemMode),
    ThresholdExceeded { temp: i16 },
    ButtonPressed,
}

// ============ 通信原语实例 ============

// 传感器数据流：采集 → 处理（Channel，不丢数据）
static SENSOR_CHAN: Channel<CriticalSectionRawMutex, SensorReading, 8> = Channel::new();

// 系统模式：配置任务 → 多个消费者（Watch）
static MODE: Watch<CriticalSectionRawMutex, SystemMode> = Watch::new();

// 事件广播：→ LED、日志、BLE（PubSubChannel）
static EVENTS: PubSubChannel<CriticalSectionRawMutex, SystemEvent, 8, 3, 2> = PubSubChannel::new();

// 按键通知：中断 → 任务（Signal）
static BUTTON: Signal<CriticalSectionRawMutex, ()> = Signal::new();

// ============ 任务实现 ============

/// 传感器采集任务：100ms 周期
#[embassy_executor::task]
async fn sensor_task() {
    let mut sensor = init_sensor().await;
    loop {
        let reading = sensor.read().await;
        SENSOR_CHAN.send(reading).await;
        Timer::after_millis(100).await;
    }
}

/// 数据处理任务：滤波 + 阈值检测
#[embassy_executor::task]
async fn process_task() {
    let mut filter = MovingAverage::new(10);
    let mut mode_reader = MODE.reader();
    
    loop {
        let reading = SENSOR_CHAN.receive().await;
        let filtered_temp = filter.update(reading.temperature);
        
        // 阈值检测
        if filtered_temp > 850 {  // 85.0°C
            let _ = EVENTS.try_publish(SystemEvent::ThresholdExceeded {
                temp: filtered_temp,
            });
        }
        
        // 低功耗模式下降低处理频率
        if *mode_reader.get().await == SystemMode::LowPower {
            Timer::after_millis(500).await;
        }
    }
}

/// LED 指示任务：订阅事件
#[embassy_executor::task]
async fn led_task(mut led: Output<'static, PC13>) {
    let mut sub = EVENTS.subscriber().unwrap();
    loop {
        match sub.next_message().await {
            embassy_sync::pubsub::Message::Payload(SystemEvent::ThresholdExceeded { .. }) => {
                // 报警：快闪
                for _ in 0..5 {
                    led.toggle();
                    Timer::after_millis(100).await;
                }
            }
            embassy_sync::pubsub::Message::Payload(SystemEvent::ModeChanged(SystemMode::LowPower)) => {
                led.set_low();  // 低功耗：灭灯
            }
            _ => {}
        }
    }
}

/// 按键处理任务
#[embassy_executor::task]
async fn button_task() {
    loop {
        BUTTON.wait().await;
        
        // 切换模式
        let current = MODE.get_cloned();
        let next = match current {
            SystemMode::Normal => SystemMode::LowPower,
            SystemMode::LowPower => SystemMode::Normal,
            SystemMode::Alarm => SystemMode::Normal,
        };
        MODE.send(next);
        let _ = EVENTS.try_publish(SystemEvent::ModeChanged(next));
    }
}

/// 按键中断
#[interrupt]
fn EXTI0() {
    BUTTON.signal(());
}

// ============ 主入口 ============

#[embassy_executor::main]
async fn main(spawner: Spawner) {
    let p = embassy_stm32::init(Default::default());
    
    // 初始化 Watch 的初始值
    MODE.send(SystemMode::Normal);
    
    // 启动所有任务
    spawner.spawn(sensor_task()).unwrap();
    spawner.spawn(process_task()).unwrap();
    spawner.spawn(led_task(Output::new(p.PC13, Level::Low, Speed::Low))).unwrap();
    spawner.spawn(button_task()).unwrap();
    
    info!("System started!");
}

rust

架构分析#

                    ┌─────────────────────────────────────────┐
                    │              EVENTS (PubSub)             │
                    │  ┌─────────┐  ┌─────────┐  ┌────────┐  │
                    │  │LED Task │  │Log Task │  │BLE Task│  │
                    │  └─────────┘  └─────────┘  └────────┘  │
                    └──────────────────▲──────────────────────┘
                                       │ publish
┌──────────┐  Channel  ┌──────────┐   │
│ Sensor   │──────────▶│ Process  │───┘
│ Task     │           │ Task     │
└──────────┘           └──────────┘
                            ▲
                            │ Watch (mode)
                       ┌────┴─────┐
                       │ Button   │◀── Signal ── EXTI0 IRQ
                       │ Task     │
                       └──────────┘

plaintext

这个架构体现了 embassy-sync 的设计哲学：

数据流用 Channel：保证每条传感器数据都被处理
状态广播用 Watch：模式变更通知所有相关任务
事件通知用 PubSub：多个独立消费者各自处理
中断通知用 Signal：最轻量的中断到任务通信
没有全局可变状态：所有共享都通过类型安全的原语

10.13 本章小结#

本章介绍了 embassy-sync 库的全部核心通信原语。让我们回顾关键要点：

核心原则#

优先使用消息传递，而非共享状态——Channel/Signal/PubSub 优于 Mutex
选择语义匹配的原语——不要试图用 Channel 模拟所有通信模式
持有 Mutex 时绝不 .await——这是异步编程的铁律
为所有等待设置超时——防止系统因某个异常而永久挂起

从 C 到 Embassy 的思维转变#

C 的思维	Embassy 的思维
全局变量 + 标志位	Signal / Watch
回调函数	PubSubChannel 订阅者
环形缓冲区 + 信号量	Channel / Pipe
`xSemaphoreTake/Give`	`Mutex.lock().await`（RAII 自动释放）
中断中调用 `FromISR` API	中断中调用 `signal()` / `try_send()`
手动管理”谁负责释放”	编译器通过所有权自动管理

下一章预告#

第 11 章将把视野从 embassy-sync 扩展到整个 Embassy 生态：USB 设备栈、网络协议栈、安全 Bootloader、蓝牙——Embassy 能做的远不止任务调度。

Comment seems to stuck. Try to refresh?✨

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

10.1 为什么需要任务间通信？#

任务隔离：Embassy 的设计哲学#

C 工程师的”老办法”及其问题#

Embassy 的答案：类型安全的通信原语#

理解 M：互斥锁类型参数#

10.2 Signal——单生产者单消费者信号#

语义：最新值覆盖旧值#

与 C 的 volatile 标志位对比#

API 速览#

适用场景#

不适用场景#

10.3 Watch——多消费者观察通道#

语义：多个读者观察同一个值的变化#

与 C 的”多个中断读取同一个全局变量”对比#

关键概念：Reader#

适用场景#

10.4 Channel——MPMC 有界队列#

语义：生产者-消费者队列#

与 FreeRTOS Queue 的对比#

关键差异#

API 速览#

中断中使用 Channel#

适用场景#

10.5 PriorityChannel——优先级通道#

语义：高优先级消息插队#

工作原理#

适用场景#

注意事项#

10.6 PubSubChannel——发布-订阅通道#

语义：一对多广播，每个订阅者独立消费#

与 C 的回调函数列表对比#

关键行为：慢订阅者#

适用场景#

10.7 Mutex——异步互斥锁#

与 C 的互斥锁对比#

⚠️ 异步 Mutex 的黄金规则：持有锁时不要 .await#

为什么 Embassy 的 Mutex 不像 tokio 那样”检测”死锁？#

RefCell 替代方案：单 Executor 场景#

适用场景#

10.8 Pipe——异步字节管道#

语义：面向字节流的异步管道#

与 C 的环形缓冲区对比#

API 速览#

适用场景#

10.9 与 FreeRTOS 通信原语的完整对比表#

一个完整的对比示例#

10.10 常见死锁与饥饿问题#

死锁模式 1：持有 Mutex 时 .await#

死锁模式 2：嵌套锁#

死锁模式 3：Channel 循环依赖#

饥饿问题#

最佳实践清单#

10.11 选型指南#

决策表#

性能与资源开销对比#

组合使用模式#

10.12 实战：完整的多任务传感器系统#

架构分析#

10.13 本章小结#

核心原则#

从 C 到 Embassy 的思维转变#

下一章预告#

理解 `M`：互斥锁类型参数#

10.2 `Signal`——单生产者单消费者信号#

与 C 的 `volatile` 标志位对比#

10.3 `Watch`——多消费者观察通道#

10.4 `Channel`——MPMC 有界队列#

10.5 `PriorityChannel`——优先级通道#

10.6 `PubSubChannel`——发布-订阅通道#

10.7 `Mutex`——异步互斥锁#

⚠️ 异步 Mutex 的黄金规则：持有锁时不要 `.await`#

`RefCell` 替代方案：单 Executor 场景#

10.8 `Pipe`——异步字节管道#

死锁模式 1：持有 Mutex 时 `.await`#