Rust 并发编程
线程、`Send`/`Sync`、`Mutex`、channel、以及 `rayon` 数据并行。
创建线程
use std::thread;
use std::time::Duration;
fn main() {
let handle = thread::spawn(|| {
for i in 1..=3 {
println!("子线程 {}", i);
thread::sleep(Duration::from_millis(10));
}
});
for i in 1..=2 {
println!("主线程 {}", i);
}
handle.join().unwrap(); // 等子线程结束
println!("完成");
}主线程 1
子线程 1
主线程 2
子线程 2
子线程 3
完成move 与所有权
use std::thread;
fn main() {
let data = vec![1, 2, 3];
// 不加 move 编译错误：闭包可能比 data 活得久
let h = thread::spawn(move || {
println!("{:?}", data);
data.len()
});
println!("{}", h.join().unwrap());
}[1, 2, 3]
3move hands ownership of data to the thread, so the compiler knows who frees it and there is no dangling read after the main thread returns.
Mutex
use std::sync::{Arc, Mutex};
use std::thread;
fn main() {
let counter = Arc::new(Mutex::new(0));
let mut handles = vec![];
for _ in 0..10 {
let c = Arc::clone(&counter);
handles.push(thread::spawn(move || {
let mut n = c.lock().unwrap();
*n += 1;
}));
}
for h in handles {
h.join().unwrap();
}
println!("{}", *counter.lock().unwrap());
}10Arc is an atomic reference count enabling shared ownership; Mutex serialises access. Arc<Mutex<T>> is the standard shared-state pattern in Rust.
| Type | Purpose |
|---|---|
Arc<T> | 多线程共享所有权 |
Rc<T> | 单线程共享所有权（更轻） |
Mutex<T> | 互斥锁，独占访问 |
RwLock<T> | 读多写少，多读并行 |
AtomicUsize | 简单计数，无锁 |
channel
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();
for i in 0..3 {
let tx = tx.clone();
thread::spawn(move || {
tx.send(i * 10).unwrap();
});
}
drop(tx); // 关掉主发送端，否则 rx 永远不结束
for v in rx {
println!("收到 {}", v);
}
}收到 0
收到 10
收到 20mpsc is multi-producer single-consumer. Iterating rx blocks until every sender is dropped — forgetting drop(tx) is a classic deadlock.
Send 与 Sync
// Send：所有权可以跨线程转移
// Sync：&T 可以跨线程共享
fn assert_send<T: Send>() {}
fn assert_sync<T: Sync>() {}
fn main() {
assert_send::<i32>();
assert_send::<String>();
assert_send::<Vec<u8>>();
assert_sync::<i32>();
assert_sync::<String>();
// Rc 不是 Send：引用计数不是原子的
// assert_send::<std::rc::Rc<i32>>(); // 编译错误
}（编译通过，无输出）Rc<T> is not Send — its count is non-atomic, so concurrent clones would miscount. This is exactly how Rust blocks data races at compile time: the type system is the lock.
数据并行
[dependencies]
rayon = "1"use rayon::prelude::*;
fn main() {
let v: Vec<u64> = (1..=1_000_000).collect();
let sum: u64 = v.par_iter().map(|x| x * 2).sum();
println!("{}", sum);
let mut nums: Vec<i32> = vec![5, 2, 8, 1, 9];
nums.par_sort();
println!("{:?}", nums);
}1000001000000
[1, 2, 5, 8, 9]Swapping iter() for par_iter() parallelises it; rayon handles chunking and scheduling. Because the borrow checker rules out data races, this "change one word" parallelism is safe.
unsafe you rarely wonder about races — at the price of verbose Arc<Mutex<>>.创建线程
use std::thread;
use std::time::Duration;
fn main() {
let handle = thread::spawn(|| {
for i in 1..=3 {
println!("子线程 {}", i);
thread::sleep(Duration::from_millis(10));
}
});
for i in 1..=2 {
println!("主线程 {}", i);
}
handle.join().unwrap(); // 等子线程结束
println!("完成");
}主线程 1
子线程 1
主线程 2
子线程 2
子线程 3
完成move 与所有权
use std::thread;
fn main() {
let data = vec![1, 2, 3];
// 不加 move 编译错误：闭包可能比 data 活得久
let h = thread::spawn(move || {
println!("{:?}", data);
data.len()
});
println!("{}", h.join().unwrap());
}[1, 2, 3]
3move 把 data 的所有权交给线程。这样编译器就确定了「谁负责释放」，不会出现主线程提前释放、子线程还在用的悬垂问题。
Mutex
use std::sync::{Arc, Mutex};
use std::thread;
fn main() {
let counter = Arc::new(Mutex::new(0));
let mut handles = vec![];
for _ in 0..10 {
let c = Arc::clone(&counter);
handles.push(thread::spawn(move || {
let mut n = c.lock().unwrap();
*n += 1;
}));
}
for h in handles {
h.join().unwrap();
}
println!("{}", *counter.lock().unwrap());
}10Arc 是原子引用计数，让多个线程共享所有权；Mutex 保证同一时刻只有一个线程能访问。这个组合 Arc<Mutex<T>> 是 Rust 并发里最常见的共享状态写法。
| 类型 | 用途 |
|---|---|
Arc<T> | 多线程共享所有权 |
Rc<T> | 单线程共享所有权（更轻） |
Mutex<T> | 互斥锁，独占访问 |
RwLock<T> | 读多写少，多读并行 |
AtomicUsize | 简单计数，无锁 |
channel
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();
for i in 0..3 {
let tx = tx.clone();
thread::spawn(move || {
tx.send(i * 10).unwrap();
});
}
drop(tx); // 关掉主发送端，否则 rx 永远不结束
for v in rx {
println!("收到 {}", v);
}
}收到 0
收到 10
收到 20mpsc 是「多生产者单消费者」。rx 用 for 迭代会一直阻塞到所有发送端都被 drop。忘了 drop(tx) 是死锁的常见原因。
Send 与 Sync
// Send：所有权可以跨线程转移
// Sync：&T 可以跨线程共享
fn assert_send<T: Send>() {}
fn assert_sync<T: Sync>() {}
fn main() {
assert_send::<i32>();
assert_send::<String>();
assert_send::<Vec<u8>>();
assert_sync::<i32>();
assert_sync::<String>();
// Rc 不是 Send：引用计数不是原子的
// assert_send::<std::rc::Rc<i32>>(); // 编译错误
}（编译通过，无输出）Rc<T> 不是 Send，因为它用非原子计数，两个线程同时 clone 会数错。这正是 Rust 在编译期挡住数据竞争的方式：类型系统就是锁。
数据并行
[dependencies]
rayon = "1"use rayon::prelude::*;
fn main() {
let v: Vec<u64> = (1..=1_000_000).collect();
let sum: u64 = v.par_iter().map(|x| x * 2).sum();
println!("{}", sum);
let mut nums: Vec<i32> = vec![5, 2, 8, 1, 9];
nums.par_sort();
println!("{:?}", nums);
}1000001000000
[1, 2, 5, 8, 9]把 iter() 换成 par_iter() 就能并行，rayon 自动分块和调度。因为借用检查器保证了没有数据竞争，这种「改一个词就并行」在 Rust 里是安全的。
unsafe 之外你几乎不用怀疑「是不是有竞争」。代价是 Arc<Mutex<>> 写起来啰嗦。