Rust的并发编程格言是"无畏并发"（Fearless Concurrency）。通过所有权系统，Rust在编译期就能防止数据竞争，让你放心地编写并发代码。
学习目标：掌握线程创建与管理、通道通信、Mutex/RwLock互斥、Send/Sync特征
use std::thread;
use std::time::Duration;
fn main() {
// 基本线程创建
let handle = thread::spawn(|| {
for i in 1..=5 {
println!("子线程: 第{}次", i);
thread::sleep(Duration::from_millis(100));
}
});
for i in 1..=3 {
println!("主线程: 第{}次", i);
thread::sleep(Duration::from_millis(100));
}
// 等待线程完成
handle.join().unwrap();
println!("线程执行完毕");
// move闭包捕获变量
let data = vec![1, 2, 3];
let handle = thread::spawn(move || {
println!("子线程获得数据: {:?}", data); // data被移动到子线程
});
handle.join().unwrap();
// println!("{:?}", data); // ❌ data已被移动
// 获取线程返回值
let handle = thread::spawn(|| {
let sum: i32 = (1..=100).sum();
sum
});
let result = handle.join().unwrap();
println!("1-100的和: {}", result);
}✅ 验证通过
Rust的并发哲学："不要通过共享内存来通信，而要通过通信来共享内存"。
use std::sync::mpsc; // multiple producer, single consumer
use std::thread;
use std::time::Duration;
fn main() {
// 创建通道
let (tx, rx) = mpsc::channel();
// 发送端移到子线程
thread::spawn(move || {
let messages = vec![
"你好",
"从",
"子线程",
"发来的消息",
];
for msg in messages {
tx.send(msg.to_string()).unwrap();
thread::sleep(Duration::from_millis(200));
}
});
// 接收端在主线程
for received in rx {
println!("收到: {}", received);
}
// 多生产者
let (tx2, rx2) = mpsc::channel();
let tx2_clone = tx2.clone(); // 克隆发送端
thread::spawn(move || {
let msgs = vec!["线程A: 1", "线程A: 2"];
for m in msgs { tx2.send(m.to_string()).unwrap(); }
});
thread::spawn(move || {
let msgs = vec!["线程B: 1", "线程B: 2"];
for m in msgs { tx2_clone.send(m.to_string()).unwrap(); }
});
// 接收指定数量
for _ in 0..4 {
println!("多生产者收到: {}", rx2.recv().unwrap());
}
}✅ 验证通过
use std::sync::{Arc, Mutex};
use std::thread;
fn main() {
// Mutex保证同一时间只有一个线程能访问数据
let counter = Arc::new(Mutex::new(0));
let mut handles = vec![];
for _ in 0..10 {
let counter = Arc::clone(&counter);
let handle = thread::spawn(move || {
let mut num = counter.lock().unwrap(); // 获取锁
*num += 1; // 临界区：修改共享数据
// 锁在num离开作用域时自动释放
});
handles.push(handle);
}
for handle in handles {
handle.join().unwrap();
}
println!("结果: {}", *counter.lock().unwrap()); // 10
// Mutex内部可变性
let data = Arc::new(Mutex::new(vec![]));
let mut handles = vec![];
for i in 0..3 {
let data = Arc::clone(&data);
handles.push(thread::spawn(move || {
let mut d = data.lock().unwrap();
d.push(i);
}));
}
for h in handles { h.join().unwrap(); }
println!("数据: {:?}", *data.lock().unwrap());
}✅ 验证通过
use std::sync::{Arc, RwLock};
use std::thread;
fn main() {
// RwLock: 多个读者或一个写者
let data = Arc::new(RwLock::new(vec![1, 2, 3]));
let mut handles = vec![];
// 多个读者
for i in 0..3 {
let data = Arc::clone(&data);
handles.push(thread::spawn(move || {
let r = data.read().unwrap(); // 读锁（可多个同时持有）
println!("读者{}: {:?}", i, *r);
}));
}
// 一个写者
{
let data = Arc::clone(&data);
handles.push(thread::spawn(move || {
let mut w = data.write().unwrap(); // 写锁（独占）
w.push(4);
println!("写者: 推入了4");
}));
}
for h in handles { h.join().unwrap(); }
println!("最终数据: {:?}", *data.read().unwrap());
}✅ 验证通过
use std::sync::{Arc, Mutex};
use std::thread;
fn is_prime(n: u64) -> bool {
if n < 2 { return false; }
if n < 4 { return true; }
if n % 2 == 0 || n % 3 == 0 { return false; }
let mut i = 5;
while i * i <= n {
if n % i == 0 || n % (i + 2) == 0 { return false; }
i += 6;
}
true
}
fn parallel_prime_count(range_start: u64, range_end: u64, num_threads: usize) -> Vec {
let primes = Arc::new(Mutex::new(Vec::new()));
let chunk_size = (range_end - range_start) / num_threads as u64;
let handles: Vec<_> = (0..num_threads).map(|i| {
let primes = Arc::clone(&primes);
let start = range_start + i as u64 * chunk_size;
let end = if i == num_threads - 1 { range_end } else { start + chunk_size };
thread::spawn(move || {
let mut local_primes = Vec::new();
for n in start..end {
if is_prime(n) {
local_primes.push(n);
}
}
primes.lock().unwrap().extend(local_primes);
})
}).collect();
for h in handles { h.join().unwrap(); }
let mut result = Arc::try_unwrap(primes).unwrap().into_inner().unwrap();
result.sort();
result
}
fn main() {
let start = 2u64;
let end = 10_000u64;
println!("🔍 查找{}-{}之间的素数...", start, end);
let primes = parallel_prime_count(start, end, 4);
println!("找到{}个素数", primes.len());
println!("前20个: {:?}", &primes[..20]);
println!("最后5个: {:?}", &primes[primes.len()-5..]);
// 验证：单线程对比
let single_count = (start..end).filter(|&n| is_prime(n)).count();
println!("验证（单线程）: {}个素数 ✅ 匹配!", single_count);
} ✅ 验证通过
实现简单线程池：创建N个工作线程，通过通道分发任务，收集结果。
实现多生产者多消费者模型：多个线程生产随机数，多个线程消费并计算平均值。
实现parallel_map<T, U>(data: Vec<T>, f: impl Fn(T) -> U) -> Vec<U>，将计算分配到多线程。
🔒 下一课解锁：async/await异步 —— 高效的异步编程
use std::sync::{Arc, Mutex, mpsc};
use std::thread;
// 线程池模式
struct ThreadPool {
workers: Vec>>,
sender: Option>>,
}
impl ThreadPool {
fn new(size: usize) -> Self {
let (sender, receiver) = mpsc::channel();
let receiver = Arc::new(Mutex::new(receiver));
let mut workers = Vec::with_capacity(size);
for _ in 0..size {
let receiver = Arc::clone(&receiver);
workers.push(Some(thread::spawn(move || {
loop {
let job = receiver.lock().unwrap().recv();
match job {
Ok(job) => job(),
Err(_) => break,
}
}
})));
}
ThreadPool { workers, sender: Some(sender) }
}
fn execute(&self, f: F)
where F: FnOnce() + Send + 'static {
self.sender.as_ref().unwrap().send(Box::new(f)).unwrap();
}
}
impl Drop for ThreadPool {
fn drop(&mut self) {
drop(self.sender.take());
for worker in &mut self.workers {
if let Some(handle) = worker.take() {
handle.join().unwrap();
}
}
}
}
fn main() {
let pool = ThreadPool::new(4);
for i in 0..8 {
pool.execute(move || {
println!("任务{}在线程{:?}执行", i, thread::current().id());
});
}
drop(pool);
println!("所有任务完成");
} ✅ 验证通过
use std::sync::{Arc, Mutex};
use std::thread;
fn parallel_map_reduce(
data: Vec,
map_fn: F,
reduce_fn: G,
num_threads: usize,
) -> U
where
T: Send + 'static,
U: Send + 'static,
F: Fn(T) -> U + Send + Sync + 'static,
G: Fn(Vec) -> U + Send + 'static,
{
let chunk_size = (data.len() + num_threads - 1) / num_threads;
let map_fn = Arc::new(map_fn);
let results = Arc::new(Mutex::new(Vec::new()));
let handles: Vec<_> = data.into_iter()
.collect::>()
.chunks(chunk_size)
.map(|chunk| chunk.to_vec())
.map(|chunk| {
let map_fn = Arc::clone(&map_fn);
let results = Arc::clone(&results);
thread::spawn(move || {
let mapped: Vec = chunk.into_iter().map(|x| map_fn(x)).collect();
results.lock().unwrap().extend(mapped);
})
}).collect();
for h in handles { h.join().unwrap(); }
let results = Arc::try_unwrap(results).unwrap().into_inner().unwrap();
reduce_fn(results)
}
fn main() {
let data: Vec = (1..=100).collect();
let sum = parallel_map_reduce(
data,
|x| x * x,
|v| v.into_iter().sum(),
4,
);
println!("1²+2²+...+100² = {}", sum);
// 验证: n(n+1)(2n+1)/6 = 100*101*201/6 = 338350
assert_eq!(sum, 338350);
println!("✅ 并行MapReduce验证通过");
} ✅ 验证通过