ZH
PortuguêsFrançaisEspañolTiếng Việt简体中文한국어日本語EnglishDeutsch
首页 / 编程语言 / Rust

Rust 中使用 Mutex 和 channel 实现安全并发

Cris The Dev · 2026/10/28 · 10 分钟阅读 · 47

1. 概述

Rust 将自己的能力称为“fearless concurrency”（无惧并发）。编译器在构建时就会阻止数据竞争，而不是等到程序运行时才出现问题。本文介绍标准库中的两个主要工具：Mutex（配合 Arc）用于共享状态，以及 mpsc channel 用于在线程间传递消息。读完本文后，你将知道如何使用每一种工具、为它们编写测试，并避免常见的陷阱，比如死锁、poisoning 和锁持有过长等问题。

所有示例仅使用标准库，无需额外的依赖。你可以使用 cargo new orders-demo 创建项目，然后将代码粘贴到 src/main.rs。

2. 问题：在线程间共享数据

假设你有多个线程共同处理订单，想要统计订单总数。最直白的写法如下：

use std::thread;

fn main() {
    let mut total_orders = 0;

    let handles: Vec<_> = (0..4)
        .map(|_| thread::spawn(|| total_orders += 1))
        .collect();

    for h in handles {
        h.join().unwrap();
    }
}

这段代码无法编译：

error[E0373]: closure may outlive the current function, but it borrows `total_orders`, which is owned by the current function

编译器给出了两个原因：

  • thread::spawn 要求闭包是 'static。子线程可能比 main 活得更久，所以它不能借用本地变量。
  • 即使加上 move，每个线程也会得到 total_orders 的各自副本。没有人真正在修改一个共享的变量。

要解决这个问题，你需要两样东西：通过 Arc 共享所有权和通过 Mutex 同步访问。

3. 使用 Arc<Mutex<T>> 共享状态

3.1. 从多个线程统计订单

use std::sync::{Arc, Mutex};
use std::thread;

fn main() {
    let total_orders = Arc::new(Mutex::new(0u32));
    let mut handles = Vec::new();

    for _ in 0..4 {
        let counter = Arc::clone(&total_orders);
        handles.push(thread::spawn(move || {
            for _ in 0..1_000 {
                *counter.lock().unwrap() += 1;
            }
        }));
    }

    for h in handles {
        h.join().unwrap();
    }

    println!("订单总数: {}", *total_orders.lock().unwrap());
}
订单总数: 4000

逐部分解释：

  • Arc::new(Mutex::new(0u32))：Arc（原子引用计数）允许多个线程共享所有权。Mutex 确保在任何时刻只有一个线程能访问内部的值。
  • Arc::clone(&total_orders) 仅增加引用计数，不复制数据。每个线程通过 move 获得自己的 Arc。
  • counter.lock() 阻止当前线程，直到获得锁。函数返回 Result<MutexGuard<u32>, PoisonError<…>>。
  • MutexGuard 解引用为 &mut u32，所以 *... += 1 直接修改值。Guard 在语句末尾被 drop，锁自动被释放。

在 Rust 中，你不可能“忘记解锁”。你也不能在不获得锁的情况下读取值，因为数据在 Mutex 内部。

注意： 你不能用 Rc 替代 Arc。Rc 没有实现 Send，编译器在把它传给 thread::spawn 时就会报错。类似地，你不能用 RefCell 替代 Mutex，因为 RefCell 不是 Sync。

3.2. MutexGuard 的作用域

锁在 guard 被 drop 之前一直被持有。如果你把 guard 赋给一个变量，锁会保持到该变量的作用域结束：

use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Duration;

fn reserve(inventory: &Mutex<HashMap<String, u32>>, sku: &str) -> bool {
    let reserved = {
        let mut stock = inventory.lock().unwrap();
        match stock.get_mut(sku) {
            Some(qty) if *qty > 0 => {
                *qty -= 1;
                true
            }
            _ => false,
        }
    }; // guard 在这里被 drop，锁被释放

    if reserved {
        // 模拟调用缓慢的 PaymentService，不持有锁
        thread::sleep(Duration::from_millis(50));
    }
    reserved
}

fn main() {
    let inventory = Arc::new(Mutex::new(HashMap::from([("AO-THUN".to_string(), 2)])));

    let handles: Vec<_> = (0..3)
        .map(|i| {
            let inv = Arc::clone(&inventory);
            thread::spawn(move || println!("顾客 {i}: {}", reserve(&inv, "AO-THUN")))
        })
        .collect();

    for h in handles {
        h.join().unwrap();
    }
}
顾客 0: true
顾客 1: true
顾客 2: false

各行的顺序在不同运行中可能不同。结果总是两个 true 和一个 false。

代码块 { ... } 将持锁时间限制在最小值。缓慢的部分（调用支付服务）在锁被释放后运行，所以其他线程不必等待。如果需要更早地释放，你可以显式调用 drop(stock)。

3.3. Mutex 被 poison

如果一个线程在持有锁时 panic，Rust 会将 Mutex 标记为 poisoned。原因是数据可能处于不完整的状态：

use std::sync::{Arc, Mutex};
use std::thread;

fn main() {
    let balance = Arc::new(Mutex::new(100i64));

    let b = Arc::clone(&balance);
    let _ = thread::spawn(move || {
        let mut guard = b.lock().unwrap();
        *guard -= 30;
        panic!("PaymentService 在中途出错");
    })
    .join();

    match balance.lock() {
        Ok(v) => println!("余额: {}", *v),
        Err(poisoned) => {
            let v = poisoned.into_inner();
            println!("Mutex 被 poison，当前余额: {}", *v);
        }
    }
}
thread '<unnamed>' panicked at src/main.rs:10:9:
PaymentService 在中途出错
Mutex 被 poison，当前余额: 70

所以 lock() 后面的 .unwrap() 实际上的意思是“如果其他线程在持有锁时 panic 了就也 panic”。对大多数应用来说，这是合理的选择。如果你知道如何恢复数据，into_inner() 会返回 guard，让你检查和修复。

4. 使用 channel 传递消息

Mutex 回答的是“多个线程修改同一数据”。Channel 采用相反的方向：每个数据在某一时刻只有一个所有者，所有权通过 channel 传递。在很多问题中，这种方式更简单、更不容易出错。

4.1. 多生产者，单消费者

std::sync::mpsc 是 multiple producer, single consumer 的缩写：

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

#[derive(Debug)]
struct Order {
    id: u32,
    customer: String,
    amount: u64,
}

fn main() {
    let (tx, rx) = mpsc::channel::<Order>();

    for shop in 1..=3 {
        let tx = tx.clone();
        thread::spawn(move || {
            let order = Order {
                id: shop * 100,
                customer: format!("khach-{shop}"),
                amount: 150_000 * shop as u64,
            };
            tx.send(order).unwrap();
        });
    }
    drop(tx); // 重要：释放原始发送端

    let mut revenue = 0;
    for order in rx {
        println!("收到订单 #{} 来自 {}", order.id, order.customer);
        revenue += order.amount;
    }
    println!("收入: {revenue}");
}
收到订单 #100 来自 khach-1
收到订单 #300 来自 khach-3
收到订单 #200 来自 khach-2
收入: 900000

解释：

  • mpsc::channel() 返回一对 (Sender, Receiver)。每个生产者得到一个克隆的 tx.clone()。
  • tx.send(order) move order 到 channel。之后，发送线程无法再使用 order，编译器会报错。这样两个线程永远不会修改同一个 Order。
  • for order in rx 循环直到所有 Sender 都被 drop。如果遗漏 drop(tx) 这一行，原始 sender 仍然活着在 main 中，循环会永远等待。

接收顺序取决于线程的调度。总收入总是 900000。

4.2. 有界 channel 和背压

mpsc::channel() 有一个无界的缓冲区。如果生产者比消费者快，内存会逐渐膨胀。sync_channel(n) 将缓冲区限制在 n 个元素，当缓冲区满时 send 会阻塞：

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

fn main() {
    let (tx, rx) = mpsc::sync_channel::<u32>(2);

    let producer = thread::spawn(move || {
        for id in 1..=5 {
            tx.send(id).unwrap();
            println!("已发送订单 {id}");
        }
    });

    thread::sleep(Duration::from_millis(100));
    for id in rx {
        println!("处理订单 {id}");
        thread::sleep(Duration::from_millis(20));
    }
    producer.join().unwrap();
}
已发送订单 1
已发送订单 2
处理订单 1
已发送订单 3
处理订单 2
...

生产者发送了两个订单后必须等待消费者取出一些。这是自然的背压机制，对处理数据流的 pipeline 非常有用。使用 sync_channel(0) 时，每次 send 都必须等到有人 recv（称为 rendezvous）。

5. 何时不需要 Arc：scoped thread

如果线程肯定在函数返回之前完成，thread::scope 允许你直接借用数据，无需 Arc：

use std::sync::Mutex;
use std::thread;

fn main() {
    let processed = Mutex::new(Vec::new());
    let orders = vec![101, 102, 103, 104];

    thread::scope(|s| {
        for chunk in orders.chunks(2) {
            s.spawn(|| processed.lock().unwrap().extend_from_slice(chunk));
        }
    }); // scope 内的所有线程都已 join

    let mut result = processed.into_inner().unwrap();
    result.sort();
    println!("{result:?}");
}
[101, 102, 103, 104]

有了 scope，编译器知道线程活不过 processed 和 orders。所以借用 & 是合法的。scope 结束后，into_inner() 提取数据出 Mutex，无需加锁。

6. 为并发代码编写测试

你应该把逻辑分解成纯函数以便测试。把以下代码放在 src/lib.rs：

use std::sync::{mpsc, Arc, Mutex};
use std::thread;

pub fn count_orders(workers: u32, per_worker: u32) -> u32 {
    let total = Arc::new(Mutex::new(0));
    let handles: Vec<_> = (0..workers)
        .map(|_| {
            let total = Arc::clone(&total);
            thread::spawn(move || {
                for _ in 0..per_worker {
                    *total.lock().unwrap() += 1;
                }
            })
        })
        .collect();
    handles.into_iter().for_each(|h| h.join().unwrap());
    let result = *total.lock().unwrap();
    result
}

pub fn sum_payments(payments: Vec<u64>) -> u64 {
    let (tx, rx) = mpsc::channel();
    for amount in payments {
        let tx = tx.clone();
        thread::spawn(move || tx.send(amount).unwrap());
    }
    drop(tx);
    rx.iter().sum()
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn counts_every_order_exactly_once() {
        assert_eq!(count_orders(8, 10_000), 80_000);
    }

    #[test]
    fn sums_all_payments_from_channel() {
        assert_eq!(sum_payments(vec![50_000, 120_000, 30_000]), 200_000);
    }

    #[test]
    fn empty_payments_returns_zero() {
        assert_eq!(sum_payments(vec![]), 0);
    }
}
cargo test
running 3 tests
test tests::empty_payments_returns_zero ... ok
test tests::sums_all_payments_from_channel ... ok
test tests::counts_every_order_exactly_once ... ok

test result: ok. 3 passed; 0 failed

测试 empty_payments_returns_zero 检查边界情况：如果忘记 drop(tx)，这个测试会永远挂起而不是返回 0。这是发现错误的明确标志。

count_orders 中有一个细节：结果被赋给一个变量后才返回。原因是临时 guard 在函数的最后一个表达式中可能活得过长，编译器会报错。

7. 常见错误

由于锁定顺序相反导致的死锁。 线程 A 锁定 orders 然后 customers，而线程 B 锁定 customers 然后 orders。两者都会永远等待。Rust 阻止数据竞争，不阻止死锁。你应该制定一个固定的锁定顺序约定，或者把两个数据合并到一个 Mutex<(A, B)> 中。

Guard 临时变量活得比你想象的久。 看这段代码：

while let Some(order) = queue.lock().unwrap().pop() {
    process(order); // 锁仍然被持有！
}

while let 中的临时变量（以及 match）会活到循环体结束。任何其他线程想往 queue 里推条目都会被阻塞，直到 process 完成。修复方法是分开：let next = queue.lock().unwrap().pop(); 然后再处理。

在同一线程上重复调用 lock()。 std::sync::Mutex 不可重入。在同一线程上调用 lock() 两次会导致死锁或 panic。

在持有锁时等待 I/O 或 recv()。 任何阻塞操作在持有 guard 期间都会成为 Mutex 的瓶颈。你应该复制或克隆所需的数据，释放锁，然后做慢速工作。

忘记 drop Sender。 如上所述，消费者会无限等待。

注意： 在 async 代码中（Tokio），不要让 std::sync::MutexGuard 跨越 .await。使用 tokio::sync::Mutex 或 Tokio 的 channel。另外，对于计数器这样的简单原子量，AtomicU32 远快于 Mutex<u32>。

何时选择哪一个？

  • 当多个线程需要读/写同一个长期状态时使用 Mutex，比如缓存或库存。如果读明显多于写，考虑 RwLock。
  • 当数据流从一个地方到另一个地方时使用 channel，比如订单队列、pipeline 或 worker pool。Channel 确保每个值只有一个所有者，使逻辑更容易推理。

8. 总结

Rust 把常见的并发错误变成编译错误。你不能在没有 Arc 的情况下共享数据，不能在没有 Mutex 的情况下修改数据，不能重用已通过 channel 发送的值。Arc<Mutex<T>> 适合共享状态，mpsc channel 适合在线程间转移所有权。sync_channel 增加背压，thread::scope 在线程的生命期明确时帮助去掉 Arc。编译器无法保护的是死锁、过长的 guard 活期和忘记 drop sender。这些是你需要仔细审查和编写测试的地方。

rustconcurrencymutexchannelthread
作者的应用Glowmi · Lantern NightA cozy Mid-Autumn match-3 puzzleiOS · Android

同主题文章

Rust · 2026/10/27 · 9 分钟阅读 · 53

用 Axum 在 Rust 中编写 REST API

使用 Axum 0.8 构建 REST API CRUD，包括 routing、extractor、shared state、使用 IntoResponse 处理错误以及不需要打开网络端口的处理程序测试。

Rust · 2026/10/26 · 10 分钟阅读 · 49

Rust中的智能指针：Box、Rc和Arc

何时在Rust中使用Box、Rc或Arc：heap分配、单线程内共享所有权和多线程间共享所有权，配合Weak、RefCell和Mutex。