当前位置：首页 > news >正文

Rust 并发编程实战：从 Mutex 到 Channel，数据竞争的编译期防线

news 2026/6/29 16:38:57

Rust 并发编程实战：从 Mutex 到 Channel，数据竞争的编译期防线

一、并发编程的恐惧：数据竞争为何如此难防

在 C/C++ 中，数据竞争（Data Race）是最难排查的 Bug 类型之一。两个线程同时访问同一块内存，至少一个是写操作，且没有同步机制——这就是数据竞争。它的可怕之处在于：Bug 不是每次都复现，可能跑 1000 次才出现一次，而且出现时表现症状随机，可能是段错误，可能是数据损坏，也可能是看似正常但结果错误。

Rust 从语言层面解决了这个问题。编译器的借用检查器确保：同一时刻，要么有多个不可变引用，要么只有一个可变引用。这个规则延伸到并发领域就是：同一时刻，要么有多个线程读取，要么只有一个线程写入。违反这个规则的代码，编译都过不了。

但 Rust 的并发安全不是免费的。你需要理解 Mutex、RwLock、Channel、Arc 等同步原语的使用方式和性能特征。选错同步原语，可能导致性能退化甚至死锁。

本文将深入 Rust 并发编程的核心原语，给出生产环境中的最佳实践和踩坑经验。

二、Rust 并发安全机制：从类型系统到同步原语

2.1 Send 和 Sync：编译期的并发安全保证

Rust 的并发安全建立在两个 marker trait 之上：

Send：类型的所有权可以跨线程转移。大部分类型自动实现 Send。
Sync：类型的不可变引用可以跨线程共享。即&T是 Send 的。

flowchart TD A[类型 T] --> B{T: Send?} B -->|是| C[可以将 T 移动到其他线程] B -->|否| D[只能在当前线程使用<br/>如 Rc, RefCell] A --> E{T: Sync?} E -->|是| F[多个线程可同时持有 &T<br/>如 Arc, Mutex] E -->|否| G[不能跨线程共享引用<br/>如 Cell, RefCell] C --> H[跨线程传递值] F --> I[跨线程共享只读引用] H --> J[线程安全组合: Arc&lt;Mutex&lt;T&gt;&gt;] I --> J

2.2 同步原语选择指南

原语	适用场景	性能特征	死锁风险
Mutex	读写交替，写多读少	加锁开销中等	中等
RwLock	读多写少	读锁快，写锁慢	中等
Channel	生产者-消费者模式	无锁（有界通道除外）	低
Atomic	简单计数器/标志位	最快	无
Semaphore	并发数限制	中等	低

2.3 Arc 的角色：跨线程共享所有权

Arc（Atomic Reference Counted）是并发版本的Rc。它通过原子操作维护引用计数，确保多线程间安全地共享所有权。Arc本身只提供共享读取能力，要修改数据需要配合Mutex或RwLock。

三、生产级代码：Rust 并发编程的核心模式

3.1 Mutex 模式：安全的共享可变状态

use std::sync::{Arc, Mutex}; use std::thread; /// 并发安全的计数器：用 Arc<Mutex<T>> 包装 struct ConcurrentCounter { value: Arc<Mutex<i64>>, } impl ConcurrentCounter { fn new(initial: i64) -> Self { ConcurrentCounter { value: Arc::new(Mutex::new(initial)), } } /// 原子递增：获取锁 → 修改 → 释放锁 fn increment(&self, delta: i64) -> i64 { // lock() 返回 MutexGuard，drop 时自动释放锁 let mut guard = self.value.lock().unwrap(); *guard += delta; *guard } /// 读取当前值：用 MutexGuard 的 Deref 自动解引用 fn get(&self) -> i64 { *self.value.lock().unwrap() } /// 克隆 Arc：创建新的引用指向同一份数据 fn clone_handle(&self) -> Self { ConcurrentCounter { value: Arc::clone(&self.value), } } } /// 多线程并发计数示例 fn concurrent_counting() -> i64 { let counter = ConcurrentCounter::new(0); let mut handles = Vec::new(); for _ in 0..10 { let counter = counter.clone_handle(); let handle = thread::spawn(move || { for _ in 0..1000 { counter.increment(1); } }); handles.push(handle); } // 等待所有线程完成 for handle in handles { handle.join().unwrap(); } counter.get() // 结果一定是 10000 }

3.2 Channel 模式：生产者-消费者解耦

use std::sync::mpsc; use std::thread; use std::time::Duration; /// 任务定义：生产者发送给消费者的工作单元 #[derive(Debug)] struct Task { id: u32, payload: String, } /// 任务结果：消费者处理完成后返回 #[derive(Debug)] struct TaskResult { id: u32, output: String, success: bool, } /// 多生产者-单消费者模式：适合任务分发场景 fn channel_pattern() { let (task_tx, task_rx) = mpsc::channel::<Task>(); let (result_tx, result_rx) = mpsc::channel::<TaskResult>(); // 启动消费者线程：从通道接收任务并处理 let consumer = thread::spawn(move || { while let Ok(task) = task_rx.recv() { // 处理任务：模拟耗时操作 let output = task.payload.to_uppercase(); let result = TaskResult { id: task.id, output, success: true, }; // 发送处理结果，忽略接收端已关闭的错误 if result_tx.send(result).is_err() { break; } } }); // 启动多个生产者线程：向通道发送任务 let mut producers = Vec::new(); for i in 0..3 { let tx = task_tx.clone(); let producer = thread::spawn(move || { for j in 0..5 { let task = Task { id: i * 100 + j, payload: format!("任务-{}-{}", i, j), }; // send 可能失败（消费者已退出），需处理 if tx.send(task).is_err() { break; } thread::sleep(Duration::from_millis(50)); } }); producers.push(producer); } // 重要：drop 原始的 task_tx，否则消费者永远不会退出 // 因为 channel 的 recv 在所有 sender 都 drop 后才返回 Err drop(task_tx); // 等待所有生产者完成 for producer in producers { producer.join().unwrap(); } // 等待消费者完成 consumer.join().unwrap(); // 收集结果 for result in result_rx.try_iter() { println!("结果: id={}, success={}", result.id, result.success); } }

3.3 RwLock 模式：读多写少的高效并发

use std::sync::{Arc, RwLock}; use std::thread; /// 并发缓存：读远多于写的场景用 RwLock 比 Mutex 更高效 struct ConcurrentCache<K, V> where K: Eq + std::hash::Hash + Clone, V: Clone, { data: Arc<RwLock<std::collections::HashMap<K, V>>>, } impl<K, V> ConcurrentCache<K, V> where K: Eq + std::hash::Hash + Clone, V: Clone, { fn new() -> Self { ConcurrentCache { data: Arc::new(RwLock::new(std::collections::HashMap::new())), } } /// 读取缓存：多个线程可以同时持有读锁 fn get(&self, key: &K) -> Option<V> { // read() 返回 RwLockReadGuard，允许多个读者并发 let guard = self.data.read().unwrap(); guard.get(key).cloned() } /// 写入缓存：写锁是排他的，会阻塞所有读操作 fn insert(&self, key: K, value: V) { let mut guard = self.data.write().unwrap(); guard.insert(key, value); } fn clone_handle(&self) -> Self { ConcurrentCache { data: Arc::clone(&self.data), } } }

3.4 避免死锁：锁的获取顺序

use std::sync::{Arc, Mutex}; /// 死锁的典型场景：两个线程以不同顺序获取两把锁 /// 线程 A: 先锁 alpha，再锁 beta /// 线程 B: 先锁 beta，再锁 alpha /// 结果：互相等待，永远无法继续 /// 修复方案：统一锁的获取顺序 /// 所有线程都按相同顺序获取锁，死锁不可能发生 struct OrderedLocks { /// 锁的获取顺序：永远先 alpha 后 beta alpha: Arc<Mutex<Vec<String>>>, beta: Arc<Mutex<Vec<String>>>, } impl OrderedLocks { fn new() -> Self { OrderedLocks { alpha: Arc::new(Mutex::new(Vec::new())), beta: Arc::new(Mutex::new(Vec::new())), } } /// 安全操作：按固定顺序获取两把锁 fn transfer(&self, from_alpha: bool, item: String) { // 无论业务逻辑如何，都先锁 alpha 再锁 beta let mut alpha_guard = self.alpha.lock().unwrap(); let mut beta_guard = self.beta.lock().unwrap(); if from_alpha { alpha_guard.retain(|x| x != &item); beta_guard.push(item); } else { beta_guard.retain(|x| x != &item); alpha_guard.push(item); } // guard 按 LIFO 顺序 drop：先释放 beta，再释放 alpha } }