Rust Tokio 入门：Tokio Channel：任务间通信的正确姿势

不吃草的牛德

发布于 2026-04-23 13:01:22

620

文章被收录于专栏：RustRust

前四篇我们从概念到配置，再到网络编程，已经打好了基础。今天我们来讲任务间通信 —— Tokio Channel。

在异步编程中，多个 Task 之间需要安全、高效地传递数据。很多人第一反应是“用锁”（tokio::sync::Mutex），但正确姿势是优先使用 Channel（消息传递）。

为什么？因为 Channel 符合“不要通过共享内存来通信，而要通过通信来共享内存”的设计哲学。它更安全（避免死锁）、更清晰（数据流向明确）、也更容易实现背压（backpressure）。

Tokio 在 tokio::sync 模块提供了四种主流 Channel，我们一一拆解。

一、四大 Channel 类型对比（先看总览）

Channel 类型	生产者-消费者关系	能否发送多个值	典型使用场景	是否有背压（bounded）
mpsc	多生产者 - 单消费者	是	任务池、工作队列、生产者-消费者	支持（推荐 bounded）
oneshot	单生产者 - 单消费者	仅一次	请求-响应、一次性结果返回	无（单值）
broadcast	多生产者 - 多消费者	是	聊天室、实时广播、Pub/Sub	支持（有容量限制）
watch	多生产者 - 多消费者	是（保留最新）	配置更新、状态同步、配置热更新	无（只保留最新值）

关键概念解释：

• 背压（Backpressure）：当消费者处理不过来时，生产者会被迫等待或失败，从而防止内存爆炸。
• bounded vs unbounded：bounded 有容量上限（推荐生产环境），unbounded 无限增长（风险高，仅用于已知不会失控的场景）。

二、mpsc：最常用的“工作队列”

mpsc（multi-producer single-consumer）是生产环境中使用频率最高的通道。

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    // 创建 bounded 通道，容量 32（推荐做法）
    let (tx, mut rx) = mpsc::channel::<String>(32);

    // 生产者 Task
    for i in 0..10 {
        let tx = tx.clone();  // Sender 可以 clone，支持多生产者
        tokio::spawn(async move {
            tx.send(format!("消息 #{}", i)).await.unwrap();
        });
    }
  // 关键一步：显式 drop 掉最初的 tx，否则 rx.recv() 永远不会结束,生产环境根据实际情况来决定
    drop(tx);
    // 消费者（单消费者）
    while let Some(msg) = rx.recv().await {
        println!("收到: {}", msg);
    }
}

bounded vs unbounded：

• mpsc::channel(32) → bounded，有背压，send 满时会 await（推荐）。
• mpsc::unbounded_channel() → unbounded，send 永不阻塞，但消费者慢时可能 OOM（慎用）。

实战建议：几乎所有后台 worker、任务队列都用 bounded mpsc。

三、oneshot：一次性“请求-响应”

当你只需要发送一个值（比如 RPC 调用结果、任务完成通知）时，用 oneshot 最合适。

use tokio::sync::oneshot;

async fn do_work() -> String {
    "任务完成".to_string()
}

#[tokio::main]
async fn main() {
    let (tx, rx) = oneshot::channel();

    tokio::spawn(async move {
        let result = do_work().await;
        let _ = tx.send(result);  // 发送一次后通道关闭
    });

    match rx.await {
        Ok(value) => println!("收到结果: {}", value),
        Err(_) => println!("发送方已丢弃"),
    }
}

经典用法：在 mpsc 任务中，把 oneshot Sender 随任务一起发送，实现“发请求 + 等响应”模式。

四、broadcast：真正的“广播”通道

适合聊天室、实时通知、多订阅者场景。每个接收者都能收到所有消息。

use tokio::sync::broadcast;

#[tokio::main]
async fn main() {
    let (tx, _) = broadcast::channel::<String>(16);  // 容量 16

    // 模拟多个订阅者
    for i in 0..3 {
        let mut rx = tx.subscribe();
        tokio::spawn(async move {
            while let Ok(msg) = rx.recv().await {
                println!("订阅者 {} 收到: {}", i, msg);
            }
        });
    }

    // 广播消息
    tx.send("大家好！".to_string()).unwrap();
    tx.send("新消息来了".to_string()).unwrap();
}

注意：当缓冲区满时，后续消息会丢弃旧消息，接收者可通过 recv() 错误得知跳过了多少条。 避坑指南：在聊天室场景中，broadcast 发出的消息，发送者自己也会收到。如果不做处理，屏幕上会出现“回声”。通常会判断消息来源或定义消息结构体包含 user_id。

五、watch：状态同步神器（只关心“最新值”）

watch 通道只保留最新的一个值，适合配置热更新、共享状态监视。

use tokio::sync::watch;
use tokio::time::Duration;
#[tokio::main]
async fn main() {
    let (tx, mut rx) = watch::channel("初始配置".to_string());

    // 订阅者
    tokio::spawn(async move {
        while rx.changed().await.is_ok() {
            let value = rx.borrow_and_update();
            println!("配置更新为: {}", *value);
        }
    });

    // 更新配置
    tx.send("新版本配置 v2".to_string()).unwrap();
    tx.send("新版本配置 v3".to_string()).unwrap();

    // 关键：给子任务留出被调度和打印的时间
        tokio::time::sleep(Duration::from_millis(100)).await;
}

特点：消费者调用 changed().await 等待变化，borrow() 获取最新值，不会积压历史消息。

六、实战案例：用 Channel 升级 Echo Server 为简单聊天室

结合上一期的 TCP Echo Server，我们可以用 broadcast 实现多人聊天：

• 主 Task 用 broadcast 发送消息
• 每个客户端连接 spawn Task，订阅 broadcast 并转发给客户端

// 核心片段
let (broadcast_tx, _) = broadcast::channel(100);

// 客户端处理
tokio::spawn(async move {
    let mut rx = broadcast_tx.subscribe();
    loop {
        tokio::select! {
            // 读取客户端消息
            Ok(n) = socket.read(...) => { /* 转发给 broadcast_tx */ }
            // 接收广播消息并发送给客户端
            Ok(msg) = rx.recv() => { socket.write_all(msg.as_bytes()).await; }
        }
    }
});