Learn Rust Coding · 课时

mpsc 通道

Q: 「mpsc 通道」这节课中我会学到什么？

在线程之间发送 你通过在浏览器中直接运行的动手代码来练习 Learn Rust Coding，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

在线程之间发送

mpsc 通道 是 CoddyKit 上的免费 Learn Rust Coding 课时。这是第 1 节课，共 4 节。你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。这是 Learn Rust Coding 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 Learn Rust Coding 课程共包含 4 节课。

什么是通道？

通道是一条单向管道，用于将值从一个线程发送到另一个线程。Rust 标准库提供了 std::sync::mpsc，其中 mpsc 表示多生产者、单消费者。

Sender 这一端负责推入值。
Receiver 这一端负责取出值。

通道让线程通过传递消息而不是直接共享内存来通信，从而避免许多数据竞争。

创建通道

调用 mpsc::channel() 可获得一个由 (Sender, Receiver) 组成的元组。这里会在线程启动后，将一个值从该线程发送回主线程。

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();
    thread::spawn(move || {
        tx.send(42).unwrap();
    });
    let received = rx.recv().unwrap();
    println!("Got: {}", received);
}

send 和 recv

tx.send(value) 会返回一个 Result：只有在接收端已被丢弃时才会失败。rx.recv() 会一直阻塞到值到达；当所有发送端都不存在时，它会返回 Err。

send 会将值的所有权移入通道。
recv 会在另一端取出该值的所有权。

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();
    thread::spawn(move || {
        let msg = String::from("hello from thread");
        tx.send(msg).unwrap();
    });
    let text = rx.recv().unwrap();
    println!("{}", text);
}

所有权通过通道移动

由于 send 会按值接收该值，因此发送后无法再使用它。这条编译时规则保证没有线程会继续持有已由另一个线程拥有的数据的过期引用。

下面的代码如果在 send 之后打印 msg，就会产生编译错误，因此我们只使用它一次。

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();
    thread::spawn(move || {
        let data = vec![1, 2, 3];
        tx.send(data).unwrap();
        // data is moved; cannot use it here
    });
    let v = rx.recv().unwrap();
    println!("Sum: {}", v.iter().sum::<i32>());
}

遍历接收端

Receiver 实现了 IntoIterator。遍历它会逐个产出值，直到通道关闭（所有发送端都被丢弃）。这是消费消息流的惯用方式。

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();
    thread::spawn(move || {
        for i in 1..=3 {
            tx.send(i).unwrap();
        }
    });
    for received in rx {
        println!("Received: {}", received);
    }
}

使用 clone 创建多个生产者

mpsc 中的 mp 表示可以拥有多个发送端。请 clone Sender，并将副本分别交给各个线程。接收端会收集所有值，直到每个副本都被丢弃。

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();
    let tx2 = tx.clone();
    thread::spawn(move || { tx.send("from A").unwrap(); });
    thread::spawn(move || { tx2.send("from B").unwrap(); });
    for msg in rx {
        println!("{}", msg);
    }
}

通道关闭语义

当所有发送端都被丢弃后，接收端循环会自动结束。如果哪怕只有一个 Sender 仍然存活，for msg in rx 就会永久阻塞，等待更多值。请始终正确地丢弃发送端，或将其限制在正确的作用域内，以便循环结束。

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();
    let handle = thread::spawn(move || {
        for i in 0..3 {
            tx.send(i * 10).unwrap();
        }
        // tx dropped here, closing the channel
    });
    handle.join().unwrap();
    let total: i32 = rx.iter().sum();
    println!("Total: {}", total);
}

使用 try_recv 进行非阻塞读取

recv 会阻塞，而 try_recv 会立即返回一个 Result。如果消息已就绪，它会返回 Ok(value)；如果通道为空或已断开，则返回 Err。这适用于必须继续执行其他工作的事件循环。

use std::sync::mpsc;
use std::thread;
use std::time::Duration;

fn main() {
    let (tx, rx) = mpsc::channel();
    thread::spawn(move || {
        thread::sleep(Duration::from_millis(50));
        tx.send("ready").unwrap();
    });
    loop {
        match rx.try_recv() {
            Ok(msg) => { println!("{}", msg); break; }
            Err(_) => println!("waiting..."),
        }
        thread::sleep(Duration::from_millis(20));
    }
}

发送自定义类型

任何实现了 Send 的类型都可以通过通道传输，包括您自己的结构体和枚举。枚举非常适合在工作线程协议中表示不同的消息类型。

use std::sync::mpsc;
use std::thread;

enum Job {
    Print(String),
    Add(i32, i32),
}

fn main() {
    let (tx, rx) = mpsc::channel();
    thread::spawn(move || {
        tx.send(Job::Print(String::from("hi"))).unwrap();
        tx.send(Job::Add(2, 3)).unwrap();
    });
    for job in rx {
        match job {
            Job::Print(s) => println!("print: {}", s),
            Job::Add(a, b) => println!("add: {}", a + b),
        }
    }
}

sync_channel 和背压

mpsc::sync_channel(n) 会创建一个缓冲区大小为 n 的有界通道。当缓冲区已满时，send 会阻塞，直到有空间可用。这会产生背压，防止速度较快的生产者压垮速度较慢的消费者。

sync_channel(0) 是一个会合通道：发送和接收必须同时发生。

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::sync_channel(2);
    thread::spawn(move || {
        for i in 1..=4 {
            tx.send(i).unwrap();
            println!("sent {}", i);
        }
    });
    for v in rx {
        println!("got {}", v);
    }
}

简单的工作线程模式

通道非常适合生产者/消费者模式：一个线程生成工作项，另一个线程消费并处理它们。在这里，主线程生成数字，工作线程计算每个数字的平方，然后通过第二个通道报告结果。

use std::sync::mpsc;
use std::thread;

fn main() {
    let (job_tx, job_rx) = mpsc::channel();
    let (res_tx, res_rx) = mpsc::channel();
    thread::spawn(move || {
        for n in job_rx {
            res_tx.send(n * n).unwrap();
        }
    });
    for n in 1..=4 {
        job_tx.send(n).unwrap();
    }
    drop(job_tx);
    for r in res_rx {
        println!("square: {}", r);
    }
}

快速检查

测试您对 mpsc 通道的理解。

回顾

您学习了 mpsc 通道的核心知识：

mpsc::channel() 返回一个 (Sender, Receiver) 对。
send 将值移入通道；recv 会阻塞，直到取出值。
遍历接收器会持续消费消息，直到所有发送器都被释放。
克隆 Sender 即可支持多个生产者。
try_recv 不会阻塞；sync_channel(n) 会增加有界反压。

通道通过传递所有权而不是共享内存，让线程能够安全地共享数据。

常见问题解答

「mpsc 通道」课时是免费的吗？

是的 — 「mpsc 通道」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 Learn Rust Coding 课程的其余内容，请升级到 CoddyKit PRO。 Learn Rust Coding 课程共包含 4 节课。

「mpsc 通道」这节课中我会学到什么？

在线程之间发送你通过在浏览器中直接运行的动手代码来练习 Learn Rust Coding，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 Learn Rust Coding 需要有经验吗？

无需任何先前经验。CoddyKit 上的 Learn Rust Coding 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。这是第 1 节课，共 4 节。

「mpsc 通道」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 Learn Rust Coding 课中编写并运行代码吗？

能。每节 Learn Rust Coding 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

← 返回 Learn Rust Coding