Rust并行计算完全教程提升程序性能的实用技巧

1. Rust并行计算基础

并行计算是现代程序开发中提升性能的重要手段，而Rust语言凭借其独特的内存安全特性和零成本抽象，成为并行计算的理想选择。Rust的所有权系统在编译时消除了数据竞争，使得并行编程既安全又高效。

1.1 并行与并发的区别

在深入Rust并行计算之前，我们需要明确两个概念：

并发（Concurrency）：处理多个任务的能力，任务可能交替执行，但不一定同时执行。
并行（Parallelism）：同时执行多个任务的能力，通常需要多核处理器支持。

Rust同时支持并发和并行编程，但本教程主要关注并行计算，即如何利用多核处理器同时执行计算任务以提高性能。

1.2 Rust并行计算的优势

Rust在并行计算方面具有以下优势：

内存安全：Rust的所有权系统在编译时防止数据竞争，无需垃圾回收器。
零成本抽象：Rust的高级抽象不会带来运行时性能损失。
精细控制：Rust允许开发者对并行化进行精细控制，以实现最佳性能。
丰富的生态系统：Rust拥有众多用于并行计算的库和工具。

2. Rust中的并行计算工具

Rust提供了多种并行计算工具，从标准库到第三方库，满足不同场景的需求。

2.1 标准库中的线程

Rust标准库提供了基本的线程支持，通过std::thread模块可以创建和管理线程。

use std::thread; use std::time::Duration; fn main() { // 创建一个新线程 let handle = thread::spawn(|| { println!("新线程开始"); // 模拟一些工作 thread::sleep(Duration::from_secs(2)); println!("新线程结束"); }); // 主线程继续执行 println!("主线程继续工作"); // 等待新线程完成 handle.join().unwrap(); println!("所有线程完成"); }

2.2 使用Rayon进行数据并行

Rayon是Rust中最流行的数据并行库，它提供了简单的API将顺序计算转换为并行计算。

首先，在Cargo.toml中添加Rayon依赖：

[dependencies] rayon = "1.8"

然后，可以使用Rayon的并行迭代器：

use rayon::prelude::*; fn main() { let numbers: Vec<i32> = (1..=100).collect(); // 顺序计算 let sequential_sum: i32 = numbers.iter().sum(); println!("顺序计算结果: {}", sequential_sum); // 并行计算 let parallel_sum: i32 = numbers.par_iter().sum(); println!("并行计算结果: {}", parallel_sum); // 并行映射和归约 let result: i32 = numbers.par_iter() .map(|&x| x * x) // 并行计算平方 .reduce(|| 0, |a, b| a + b); // 并行求和 println!("平方和: {}", result); }

2.3 使用Crossbeam进行高级并发

Crossbeam提供了比标准库更丰富的并发原语，适用于更复杂的并行场景。

在Cargo.toml中添加Crossbeam依赖：

[dependencies] crossbeam = "0.8"

使用Crossbeam的通道（channel）进行线程间通信：

use crossbeam_channel::{unbounded, Sender, Receiver}; use std::thread; fn main() { // 创建无界通道 let (sender, receiver) = unbounded(); // 启动多个生产者线程 let mut handles = vec![]; for i in 0..5 { let sender = sender.clone(); let handle = thread::spawn(move || { for j in 0..10 { let value = i * 10 + j; sender.send(value).unwrap(); println!("发送值: {}", value); thread::sleep(std::time::Duration::from_millis(100)); } }); handles.push(handle); } // 启动消费者线程 let consumer_handle = thread::spawn(move || { for _ in 0..50 { // 知道总共会发送50个值 let value = receiver.recv().unwrap(); println!("接收值: {}", value); } }); // 等待所有生产者完成 for handle in handles { handle.join().unwrap(); } // 等待消费者完成 consumer_handle.join().unwrap(); println!("所有操作完成"); }

3. 数据并行处理

数据并行是并行计算中最常见的模式，它将数据集分割成多个部分，然后在不同的处理器上并行处理这些部分。

3.1 并行迭代器

Rayon的并行迭代器是进行数据并行处理的简单方法。它们与标准库的迭代器API类似，但会自动利用所有可用的CPU核心。

use rayon::prelude::*; fn main() { let data: Vec<i32> = (1..=1_000_000).collect(); // 并行过滤 let even_numbers: Vec<i32> = data.par_iter() .cloned() .filter(|&x| x % 2 == 0) .collect(); println!("找到 {} 个偶数", even_numbers.len()); // 并行映射 let squares: Vec<i32> = data.par_iter() .map(|&x| x * x) .collect(); println!("计算了 {} 个平方数", squares.len()); // 并行排序 let mut unsorted = vec![5, 3, 8, 1, 9, 2, 7, 4, 6]; unsorted.par_sort(); println!("排序结果: {:?}", unsorted); }

3.2 并行归约

归约操作是将数据集合并为单个值的操作，如求和、求最大值等。Rayon提供了高效的并行归约操作。

use rayon::prelude::*; fn main() { let data: Vec<i64> = (1..=10_000_000).collect(); // 并行求和 let sum: i64 = data.par_iter().sum(); println!("总和: {}", sum); // 并行求最大值 let max = data.par_iter().max().unwrap(); println!("最大值: {}", max); // 自定义并行归约 let product = data.par_iter() .cloned() .reduce(|| 1, |a, b| a * b); println!("乘积: {}", product); // 使用fold和reduce进行更复杂的归约 let average = data.par_iter() .cloned() .fold(|| (0, 0), |(sum, count), x| (sum + x, count + 1)) .reduce(|| (0, 0), |(sum1, count1), (sum2, count2)| (sum1 + sum2, count1 + count2)); println!("平均值: {}", average.0 as f64 / average.1 as f64); }

3.3 分区操作

分区操作允许你根据条件将数据分成不同的组，这在数据分析中非常有用。

use rayon::prelude::*; use std::collections::HashMap; fn main() { let words = vec![ "apple", "banana", "cherry", "date", "elderberry", "fig", "grape", "honeydew", "kiwi", "lemon" ]; // 按首字母分区 let mut groups: HashMap<char, Vec<&str>> = HashMap::new(); words.par_iter().for_each(|&word| { let first_char = word.chars().next().unwrap(); groups.entry(first_char).or_insert_with(Vec::new).push(word); }); // 打印分组结果 for (key, values) in groups.iter_mut() { values.sort(); // 对每个组内的单词排序 println!("{}: {:?}", key, values); } }

4. 任务并行处理

任务并行关注于将不同的任务分配到不同的处理器上执行，这些任务可能处理不同的数据或执行不同的操作。

4.1 使用线程池

Rust标准库没有内置线程池，但我们可以使用Rayon的线程池或第三方库如threadpool。

使用Rayon的线程池：

use rayon::prelude::*; use std::sync::mpsc; use std::thread; fn main() { // 获取Rayon的当前线程池 let pool = rayon::ThreadPoolBuilder::new().num_threads(4).build().unwrap(); let (sender, receiver) = mpsc::channel(); // 在线程池中执行多个任务 pool.install(|| { (0..10).into_par_iter().for_each(|i| { let result = heavy_computation(i); sender.send((i, result)).unwrap(); }); }); // 关闭发送端，确保所有数据都已发送 drop(sender); // 收集结果 let mut results = Vec::new(); for (i, result) in receiver { results.push((i, result)); } // 按原始顺序排序结果 results.sort_by_key(|&(i, _)| i); // 打印结果 for (i, result) in results { println!("任务 {} 的结果: {}", i, result); } } fn heavy_computation(n: i32) -> i32 { // 模拟耗时计算 thread::sleep(std::time::Duration::from_millis(100)); n * n }

4.2 异步并行任务

使用async-std或tokio等异步运行时可以有效地并行执行异步任务。

在Cargo.toml中添加tokio依赖：

[dependencies] tokio = { version = "1.0", features = ["full"] }

使用tokio执行并行异步任务：

use tokio::time::{sleep, Duration}; #[tokio::main] async fn main() { // 创建多个异步任务 let task1 = async { println!("任务1开始"); sleep(Duration::from_secs(1)).await; println!("任务1完成"); "任务1的结果" }; let task2 = async { println!("任务2开始"); sleep(Duration::from_secs(2)).await; println!("任务2完成"); "任务2的结果" }; let task3 = async { println!("任务3开始"); sleep(Duration::from_secs(1)).await; println!("任务3完成"); "任务3的结果" }; // 并行执行所有任务 let results = tokio::join!(task1, task2, task3); println!("所有任务完成:"); println!("结果1: {}", results.0); println!("结果2: {}", results.1); println!("结果3: {}", results.2); }

4.3 使用Futures进行细粒度并行

Future是Rust中表示异步计算的概念，可以用于构建复杂的并行计算流程。

use futures::future::{join_all, FutureExt}; async fn async_task(id: usize, delay_ms: u64) -> String { println!("任务 {} 开始", id); tokio::time::sleep(tokio::time::Duration::from_millis(delay_ms)).await; println!("任务 {} 完成", id); format!("任务 {} 的结果", id) } #[tokio::main] async fn main() { // 创建多个future let futures = vec![ async_task(1, 100).boxed(), async_task(2, 200).boxed(), async_task(3, 150).boxed(), async_task(4, 50).boxed(), async_task(5, 300).boxed(), ]; // 并行执行所有future let results = join_all(futures).await; println!("所有任务完成:"); for (i, result) in results.iter().enumerate() { println!("结果 {}: {}", i + 1, result); } }

5. 共享状态管理

在并行计算中，多个线程或任务可能需要访问共享数据。Rust提供了多种安全处理共享状态的机制。

5.1 使用Arc和Mutex

Arc（原子引用计数）允许在多个线程之间共享数据所有权，而Mutex（互斥锁）确保同一时间只有一个线程可以访问数据。

use std::sync::{Arc, Mutex}; use std::thread; fn main() { // 创建一个共享计数器 let counter = Arc::new(Mutex::new(0)); let mut handles = vec![]; // 启动10个线程，每个线程增加计数器1000次 for _ in 0..10 { let counter = Arc::clone(&counter); let handle = thread::spawn(move || { for _ in 0..1000 { // 锁定互斥锁以安全地修改计数器 let mut num = counter.lock().unwrap(); *num += 1; } }); handles.push(handle); } // 等待所有线程完成 for handle in handles { handle.join().unwrap(); } // 获取最终结果 println!("最终结果: {}", *counter.lock().unwrap()); }

5.2 使用RwLock进行读写分离

当数据读多写少时，RwLock比Mutex更高效，因为它允许多个读取者同时访问数据。

use std::sync::{Arc, RwLock}; use std::thread; use std::time::Duration; fn main() { // 创建一个共享值 let value = Arc::new(RwLock::new(0)); let mut handles = vec![]; // 启动3个写入者线程 for i in 0..3 { let value = Arc::clone(&value); let handle = thread::spawn(move || { for j in 0..5 { // 获取写锁 let mut num = value.write().unwrap(); *num = i * 10 + j; println!("写入者 {} 写入值: {}", i, *num); drop(num); // 显式释放锁 thread::sleep(Duration::from_millis(100)); } }); handles.push(handle); } // 启动5个读取者线程 for i in 0..5 { let value = Arc::clone(&value); let handle = thread::spawn(move || { for _ in 0..5 { // 获取读锁 let num = value.read().unwrap(); println!("读取者 {} 读取值: {}", i, *num); drop(num); // 显式释放锁 thread::sleep(Duration::from_millis(50)); } }); handles.push(handle); } // 等待所有线程完成 for handle in handles { handle.join().unwrap(); } }

5.3 使用原子类型

对于简单的数据类型，可以使用原子操作，它们比锁更高效，因为不需要阻塞线程。

use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; use std::thread; fn main() { // 创建一个原子计数器 let counter = Arc::new(AtomicUsize::new(0)); let mut handles = vec![]; // 启动10个线程，每个线程增加计数器1000次 for _ in 0..10 { let counter = Arc::clone(&counter); let handle = thread::spawn(move || { for _ in 0..1000 { // 使用原子操作增加计数器 counter.fetch_add(1, Ordering::SeqCst); } }); handles.push(handle); } // 等待所有线程完成 for handle in handles { handle.join().unwrap(); } // 获取最终结果 println!("最终结果: {}", counter.load(Ordering::SeqCst)); }

5.4 使用通道进行消息传递

Rust鼓励通过消息传递而不是共享内存来进行并发编程，通道是实现这一点的理想工具。

use std::sync::mpsc; use std::thread; use std::time::Duration; fn main() { // 创建一个通道 let (sender, receiver) = mpsc::channel(); // 启动发送者线程 let sender_handle = thread::spawn(move || { for i in 0..10 { sender.send(i).unwrap(); println!("发送: {}", i); thread::sleep(Duration::from_millis(100)); } println!("发送完成"); }); // 启动接收者线程 let receiver_handle = thread::spawn(move || { for _ in 0..10 { let received = receiver.recv().unwrap(); println!("接收: {}", received); } println!("接收完成"); }); // 等待两个线程完成 sender_handle.join().unwrap(); receiver_handle.join().unwrap(); }

6. 性能优化技巧

并行计算的目标是提高性能，但简单地并行化代码并不总能带来性能提升。以下是一些优化Rust并行程序性能的技巧。

6.1 避免数据竞争和假共享

数据竞争会导致未定义行为，而假共享会显著降低性能。Rust在编译时防止数据竞争，但开发者仍需注意假共享问题。

use std::sync::{Arc, Mutex}; use std::thread; fn main() { // 创建一个包含多个计数器的结构体 struct Counters { counter1: u64, counter2: u64, // 添加填充以避免假共享 _padding: [u64; 14], } let counters = Arc::new(Mutex::new(Counters { counter1: 0, counter2: 0, _padding: [0; 14], })); let mut handles = vec![]; // 启动线程1，只修改counter1 let counters1 = Arc::clone(&counters); let handle1 = thread::spawn(move || { for _ in 0..1000000 { let mut counters = counters1.lock().unwrap(); counters.counter1 += 1; } }); handles.push(handle1); // 启动线程2，只修改counter2 let counters2 = Arc::clone(&counters); let handle2 = thread::spawn(move || { for _ in 0..1000000 { let mut counters = counters2.lock().unwrap(); counters.counter2 += 1; } }); handles.push(handle2); // 等待所有线程完成 for handle in handles { handle.join().unwrap(); } // 打印结果 let counters = counters.lock().unwrap(); println!("Counter1: {}", counters.counter1); println!("Counter2: {}", counters.counter2); }

6.2 减少锁竞争

锁竞争会严重限制并行程序的性能。以下是减少锁竞争的几种方法：

减少临界区大小：尽可能减少持有锁的时间。
使用读写锁：当读多写少时，使用RwLock代替Mutex。
使用无锁数据结构：对于简单操作，使用原子类型。
数据分片：将数据分成多个部分，每个部分使用不同的锁。

use std::sync::{Arc, Mutex}; use std::thread; fn main() { // 创建多个分片，每个分片有自己的锁 const NUM_SHARDS: usize = 4; let shards: Vec<_> = (0..NUM_SHARDS) .map(|_| Arc::new(Mutex::new(0))) .collect(); let mut handles = vec![]; // 启动多个线程，每个线程随机更新一个分片 for _ in 0..10 { let shards = shards.clone(); let handle = thread::spawn(move || { for i in 0..10000 { // 根据i选择分片 let shard_index = i % NUM_SHARDS; let mut shard = shards[shard_index].lock().unwrap(); *shard += 1; } }); handles.push(handle); } // 等待所有线程完成 for handle in handles { handle.join().unwrap(); } // 计算总和 let total: i32 = shards.iter().map(|shard| *shard.lock().unwrap()).sum(); println!("总和: {}", total); }

6.3 负载均衡

负载均衡是确保所有处理器都充分利用的关键。Rayon会自动处理工作窃取以实现负载均衡，但在自定义实现中需要注意这一点。

use rayon::prelude::*; use std::time::Instant; fn main() { let data: Vec<i32> = (0..1_000_000).collect(); // 使用Rayon的并行迭代器（自动负载均衡） let start = Instant::now(); let sum: i32 = data.par_iter().sum(); let duration = start.elapsed(); println!("Rayon并行求和: {}, 耗时: {:?}", sum, duration); // 手动分块实现（可能负载不均衡） let start = Instant::now(); let chunks = data.chunks(100000); // 固定大小的块 let mut partial_sums = Vec::new(); for chunk in chunks { let sum: i32 = chunk.iter().sum(); partial_sums.push(sum); } let manual_sum: i32 = partial_sums.iter().sum(); let duration = start.elapsed(); println!("手动分块求和: {}, 耗时: {:?}", manual_sum, duration); }

6.4 批处理和缓存友好性

批处理操作和考虑缓存友好性可以显著提高性能。

use rayon::prelude::*; fn main() { // 创建一个大矩阵 let size = 1000; let matrix: Vec<Vec<f64>> = (0..size) .map(|_| (0..size).map(|i| i as f64).collect()) .collect(); // 逐行处理（缓存友好） let start = std::time::Instant::now(); let row_sums: Vec<f64> = matrix.par_iter() .map(|row| row.iter().sum()) .collect(); let duration = start.elapsed(); println!("逐行处理耗时: {:?}", duration); // 逐列处理（缓存不友好） let start = std::time::Instant::now(); let col_sums: Vec<f64> = (0..size).into_par_iter() .map(|j| (0..size).map(|i| matrix[i][j]).sum()) .collect(); let duration = start.elapsed(); println!("逐列处理耗时: {:?}", duration); // 验证结果 let total_row_sum: f64 = row_sums.iter().sum(); let total_col_sum: f64 = col_sums.iter().sum(); println!("行总和: {}, 列总和: {}", total_row_sum, total_col_sum); }

7. 实际应用案例

让我们通过几个实际案例来展示Rust并行计算的应用。

7.1 图像处理

图像处理是并行计算的理想应用场景，因为像素的处理通常是独立的。

use image::{ImageBuffer, Rgb}; use rayon::prelude::*; use std::time::Instant; fn main() { // 创建一个示例图像 let width = 800; let height = 600; let mut img = ImageBuffer::new(width, height); // 填充图像 for (x, y, pixel) in img.enumerate_pixels_mut() { let r = (x as f32 / width as f32 * 255.0) as u8; let g = (y as f32 / height as f32 * 255.0) as u8; let b = 128; *pixel = Rgb([r, g, b]); } // 保存原始图像 img.save("original.png").unwrap(); // 串行灰度化 let start = Instant::now(); let mut gray_img = img.clone(); for pixel in gray_img.pixels_mut() { let gray = 0.299 * pixel[0] as f32 + 0.587 * pixel[1] as f32 + 0.114 * pixel[2] as f32; let gray_val = gray as u8; *pixel = Rgb([gray_val, gray_val, gray_val]); } let duration = start.elapsed(); gray_img.save("gray_serial.png").unwrap(); println!("串行灰度化耗时: {:?}", duration); // 并行灰度化 let start = Instant::now(); let mut parallel_gray_img = img.clone(); parallel_gray_img.par_chunks_mut(3).for_each(|chunk| { let gray = 0.299 * chunk[0] as f32 + 0.587 * chunk[1] as f32 + 0.114 * chunk[2] as f32; let gray_val = gray as u8; chunk[0] = gray_val; chunk[1] = gray_val; chunk[2] = gray_val; }); let duration = start.elapsed(); parallel_gray_img.save("gray_parallel.png").unwrap(); println!("并行灰度化耗时: {:?}", duration); }

7.2 数值计算

数值计算，如矩阵运算，是并行计算的另一个重要应用领域。

use rayon::prelude::*; use std::time::Instant; type Matrix = Vec<Vec<f64>>; fn matrix_multiply(a: &Matrix, b: &Matrix) -> Matrix { let n = a.len(); let m = b[0].len(); let p = b.len(); (0..n) .map(|i| { (0..m) .map(|j| { (0..p).map(|k| a[i][k] * b[k][j]).sum() }) .collect() }) .collect() } fn parallel_matrix_multiply(a: &Matrix, b: &Matrix) -> Matrix { let n = a.len(); let m = b[0].len(); let p = b.len(); (0..n) .into_par_iter() .map(|i| { (0..m) .map(|j| { (0..p).map(|k| a[i][k] * b[k][j]).sum() }) .collect() }) .collect() } fn main() { // 创建两个矩阵 let size = 200; let a: Matrix = (0..size) .map(|i| (0..size).map(|j| (i + j) as f64).collect()) .collect(); let b: Matrix = (0..size) .map(|i| (0..size).map(|j| (i * j) as f64).collect()) .collect(); // 串行矩阵乘法 let start = Instant::now(); let _c = matrix_multiply(&a, &b); let duration = start.elapsed(); println!("串行矩阵乘法耗时: {:?}", duration); // 并行矩阵乘法 let start = Instant::now(); let _c = parallel_matrix_multiply(&a, &b); let duration = start.elapsed(); println!("并行矩阵乘法耗时: {:?}", duration); }

7.3 Web服务

在Web服务中，并行处理请求可以显著提高吞吐量。

use actix_web::{get, web, App, HttpServer, Responder}; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; use std::time::Instant; // 共享计数器 struct AppState { counter: Arc<AtomicUsize>, } #[get("/")] async fn index(data: web::Data<AppState>) -> impl Responder { let start = Instant::now(); // 模拟一些计算工作 let _result = heavy_computation(); // 增加计数器 let count = data.counter.fetch_add(1, Ordering::SeqCst) + 1; let duration = start.elapsed(); format!( "请求 #{} 处理完成，耗时: {:?}n", count, duration ) } fn heavy_computation() -> u64 { // 模拟CPU密集型任务 (0..1_000_000).map(|i| i * i).sum() } #[actix_web::main] async fn main() -> std::io::Result<()> { // 创建共享状态 let counter = Arc::new(AtomicUsize::new(0)); println!("服务器启动在 http://localhost:8080"); HttpServer::new(move || { App::new() .data(AppState { counter: counter.clone(), }) .service(index) }) .bind("127.0.0.1:8080")? .workers(4) // 使用4个工作线程 .run() .await }

8. 常见问题和解决方案

在Rust并行计算中，开发者可能会遇到一些常见问题。本节将讨论这些问题及其解决方案。

8.1 死锁

死锁是指两个或多个线程互相等待对方释放资源，导致所有线程都无法继续执行的情况。

use std::sync::Mutex; use std::thread; fn main() { // 创建两个互斥锁 let mutex1 = Mutex::new(0); let mutex2 = Mutex::new(0); // 创建两个锁的引用 let mutex1 = &mutex1; let mutex2 = &mutex2; // 启动线程1 let handle1 = thread::spawn(move || { // 以固定顺序获取锁以避免死锁 let _lock1 = mutex1.lock().unwrap(); println!("线程1获取了锁1"); thread::sleep(std::time::Duration::from_millis(10)); let _lock2 = mutex2.lock().unwrap(); println!("线程1获取了锁2"); // 执行操作... }); // 启动线程2 let handle2 = thread::spawn(move || { // 以相同顺序获取锁 let _lock1 = mutex1.lock().unwrap(); println!("线程2获取了锁1"); thread::sleep(std::time::Duration::from_millis(10)); let _lock2 = mutex2.lock().unwrap(); println!("线程2获取了锁2"); // 执行操作... }); // 等待线程完成 handle1.join().unwrap(); handle2.join().unwrap(); println!("程序完成"); }

解决方案：

始终以相同的顺序获取多个锁。
使用try_lock而不是lock，并在失败时释放已获取的锁。
使用超时机制，避免无限等待。

8.2 过度并行化

过度并行化是指创建过多的并行任务，导致调度开销超过并行化带来的收益。

use rayon::prelude::*; use std::time::Instant; fn main() { let data: Vec<i32> = (0..1_000_000).collect(); // 适度并行化 let start = Instant::now(); let sum1: i32 = data.par_iter().sum(); let duration1 = start.elapsed(); println!("适度并行化结果: {}, 耗时: {:?}", sum1, duration1); // 过度并行化（每个元素一个任务） let start = Instant::now(); let sum2: i32 = data.into_par_iter() .map(|x| { // 每个元素都作为一个单独的任务 x }) .sum(); let duration2 = start.elapsed(); println!("过度并行化结果: {}, 耗时: {:?}", sum2, duration2); println!("性能比较: {:?} vs {:?}", duration1, duration2); }

解决方案：

使用适当的粒度，避免创建过多的小任务。
使用批处理，将多个小操作合并为一个较大的任务。
根据数据大小和计算复杂性调整并行度。

8.3 数据竞争

虽然Rust在编译时防止了大部分数据竞争，但在使用不安全代码或某些特定模式时仍可能出现。

use std::sync::Arc; use std::thread; fn main() { let mut data = vec![1, 2, 3]; let data_ptr = data.as_mut_ptr(); let handle = thread::spawn(move || { unsafe { // 这是不安全的，可能导致数据竞争 for i in 0..3 { *data_ptr.offset(i) += 1; } } }); unsafe { // 这也是不安全的，与上面的线程形成数据竞争 for i in 0..3 { *data_ptr.offset(i) *= 2; } } handle.join().unwrap(); unsafe { println!("结果: {}, {}, {}", *data_ptr.offset(0), *data_ptr.offset(1), *data_ptr.offset(2)); } }

解决方案：

避免使用不安全代码，除非绝对必要。
使用适当的同步原语，如Mutex、RwLock或原子类型。
使用Rust的安全抽象，如Arc和通道。

8.4 负载不均衡

负载不均衡是指某些处理器承担了比其他处理器更多的工作，导致整体效率降低。

use rayon::prelude::*; use std::time::Instant; fn main() { // 创建一个数据集，其中某些元素需要更多处理 let data: Vec<i32> = (0..1000).collect(); // 简单的并行处理（可能导致负载不均衡） let start = Instant::now(); let results1: Vec<i32> = data.par_iter() .map(|&x| { // 模拟不均匀的工作负载 if x % 10 == 0 { // 每10个元素中有一个需要更多工作 heavy_computation(x) } else { x * 2 } }) .collect(); let duration1 = start.elapsed(); println!("简单并行处理耗时: {:?}", duration1); // 使用分区和自定义调度（改善负载均衡） let start = Instant::now(); let (heavy_items, light_items): (Vec<_>, Vec<_>) = data.iter().partition(|&&x| x % 10 == 0); let heavy_results: Vec<i32> = heavy_items.par_iter() .map(|&&x| heavy_computation(x)) .collect(); let light_results: Vec<i32> = light_items.par_iter() .map(|&&x| x * 2) .collect(); let duration2 = start.elapsed(); println!("分区并行处理耗时: {:?}", duration2); println!("性能比较: {:?} vs {:?}", duration1, duration2); } fn heavy_computation(x: i32) -> i32 { // 模拟CPU密集型任务 (0..100000).map(|i| i * x).sum() }

解决方案：