Rust并发安全模式：从线程安全到无锁编程

第一程序员

97人浏览 · 2026-05-09 10:23:59

第一程序员 · 2026-05-09 10:23:59 发布

Rust并发安全模式：从线程安全到无锁编程

引言

并发编程是后端开发的核心挑战之一。Rust通过所有权系统和类型安全，在编译时保证并发安全，避免了数据竞争等常见问题。

本文将深入探讨Rust中的并发安全模式，包括线程同步、无锁编程、原子操作等核心技术。

一、线程安全基础

1.1 Send和Sync trait

use std::thread;

fn send_example() {
    let data = vec![1, 2, 3];
    
    // Vec<T>实现了Send，可以在线程间传递
    thread::spawn(move || {
        println!("Data: {:?}", data);
    }).join().unwrap();
}

fn sync_example() {
    let data = vec![1, 2, 3];
    let data_ref = &data;
    
    // &Vec<T>实现了Sync，可以在线程间共享引用
    thread::spawn(move || {
        println!("Data ref: {:?}", data_ref);
    }).join().unwrap();
}

1.2 线程安全的数据结构

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

fn thread_safe_hashmap() {
    let shared_map: Arc<Mutex<HashMap<String, i32>>> = Arc::new(Mutex::new(HashMap::new()));
    
    let mut handles = vec![];
    
    for i in 0..10 {
        let map = Arc::clone(&shared_map);
        let handle = thread::spawn(move || {
            let mut data = map.lock().unwrap();
            data.insert(format!("key{}", i), i);
        });
        handles.push(handle);
    }
    
    for handle in handles {
        handle.join().unwrap();
    }
    
    println!("Map size: {}", shared_map.lock().unwrap().len());
}

二、线程同步原语

2.1 Mutex和RwLock

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

fn mutex_example() {
    let counter = Mutex::new(0);
    
    let mut handles = vec![];
    
    for _ in 0..10 {
        let handle = thread::spawn(move || {
            let mut num = counter.lock().unwrap();
            *num += 1;
        });
        handles.push(handle);
    }
    
    for handle in handles {
        handle.join().unwrap();
    }
    
    println!("Result: {}", *counter.lock().unwrap());
}

fn rwlock_example() {
    let data = RwLock::new(vec![1, 2, 3]);
    
    // 多个读锁可以同时持有
    let read_handle1 = thread::spawn(|| {
        let data = data.read().unwrap();
        println!("Read 1: {:?}", data);
    });
    
    let read_handle2 = thread::spawn(|| {
        let data = data.read().unwrap();
        println!("Read 2: {:?}", data);
    });
    
    read_handle1.join().unwrap();
    read_handle2.join().unwrap();
    
    // 写锁独占
    let write_handle = thread::spawn(|| {
        let mut data = data.write().unwrap();
        data.push(4);
        println!("After write: {:?}", data);
    });
    
    write_handle.join().unwrap();
}

2.2 Condvar条件变量

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

fn condvar_example() {
    let pair = Arc::new((Mutex::new(false), Condvar::new()));
    let pair2 = Arc::clone(&pair);
    
    thread::spawn(move || {
        let (lock, cvar) = &*pair2;
        let mut started = lock.lock().unwrap();
        *started = true;
        cvar.notify_one();
        println!("Worker thread started");
    });
    
    let (lock, cvar) = &*pair;
    let mut started = lock.lock().unwrap();
    while !*started {
        started = cvar.wait(started).unwrap();
    }
    println!("Main thread detected start");
}

三、无锁编程

3.1 原子操作

use std::sync::atomic::{AtomicUsize, Ordering};
use std::thread;

fn atomic_counter() {
    static COUNTER: AtomicUsize = AtomicUsize::new(0);
    
    let mut handles = vec![];
    
    for _ in 0..1000 {
        let handle = thread::spawn(|| {
            COUNTER.fetch_add(1, Ordering::SeqCst);
        });
        handles.push(handle);
    }
    
    for handle in handles {
        handle.join().unwrap();
    }
    
    println!("Counter: {}", COUNTER.load(Ordering::SeqCst));
}

fn compare_and_swap() {
    static VALUE: AtomicUsize = AtomicUsize::new(0);
    
    let handle1 = thread::spawn(|| {
        VALUE.compare_and_swap(0, 1, Ordering::SeqCst);
    });
    
    let handle2 = thread::spawn(|| {
        VALUE.compare_and_swap(0, 2, Ordering::SeqCst);
    });
    
    handle1.join().unwrap();
    handle2.join().unwrap();
    
    println!("Value: {}", VALUE.load(Ordering::SeqCst));
}

3.2 内存顺序

use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::thread;

fn memory_ordering() {
    let ready = AtomicBool::new(false);
    let data = AtomicUsize::new(0);
    
    let producer = thread::spawn(move || {
        data.store(42, Ordering::Release);
        ready.store(true, Ordering::Release);
    });
    
    let consumer = thread::spawn(move || {
        while !ready.load(Ordering::Acquire) {}
        println!("Data: {}", data.load(Ordering::Acquire));
    });
    
    producer.join().unwrap();
    consumer.join().unwrap();
}

四、并发安全模式

4.1 生产者-消费者模式

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

fn producer_consumer() {
    let (tx, rx) = mpsc::channel();
    
    // 生产者
    let producer = thread::spawn(move || {
        for i in 0..10 {
            tx.send(i).unwrap();
            println!("Produced: {}", i);
        }
    });
    
    // 消费者
    let consumer = thread::spawn(move || {
        for received in rx {
            println!("Consumed: {}", received);
        }
    });
    
    producer.join().unwrap();
    consumer.join().unwrap();
}

fn multiple_producers() {
    let (tx, rx) = mpsc::channel();
    
    for i in 0..3 {
        let tx = tx.clone();
        thread::spawn(move || {
            tx.send(i).unwrap();
            println!("Producer {} sent: {}", i, i);
        });
    }
    
    drop(tx);
    
    for received in rx {
        println!("Consumed: {}", received);
    }
}

4.2 工作窃取模式

use crossbeam::deque::{Steal, Worker};
use std::thread;

fn work_stealing() {
    let mut workers = Vec::new();
    let mut handles = Vec::new();
    
    for i in 0..4 {
        let worker = Worker::new_fifo();
        workers.push(worker);
    }
    
    for (i, worker) in workers.iter_mut().enumerate() {
        for j in 0..10 {
            worker.push((i, j));
        }
    }
    
    for i in 0..4 {
        let workers = workers.clone();
        let handle = thread::spawn(move || {
            let mut local = Worker::new_fifo();
            
            loop {
                let mut stolen = false;
                
                for (j, worker) in workers.iter().enumerate() {
                    if j != i {
                        match worker.steal() {
                            Steal::Success(task) => {
                                println!("Thread {} stole task {:?}", i, task);
                                stolen = true;
                            }
                            Steal::Empty => continue,
                            Steal::Retry => continue,
                        }
                    }
                }
                
                if !stolen {
                    if let Some(task) = local.pop() {
                        println!("Thread {} processed local task {:?}", i, task);
                    } else {
                        break;
                    }
                }
            }
        });
        handles.push(handle);
    }
    
    for handle in handles {
        handle.join().unwrap();
    }
}

五、并发数据结构

5.1 并发安全队列

use std::sync::Arc;
use crossbeam_queue::ConcurrentQueue;

fn concurrent_queue() {
    let queue = Arc::new(ConcurrentQueue::unbounded());
    
    let mut handles = Vec::new();
    
    for i in 0..5 {
        let queue = Arc::clone(&queue);
        let handle = thread::spawn(move || {
            queue.push(i).unwrap();
            println!("Pushed: {}", i);
        });
        handles.push(handle);
    }
    
    for handle in handles {
        handle.join().unwrap();
    }
    
    while let Ok(value) = queue.pop() {
        println!("Popped: {}", value);
    }
}

5.2 无锁哈希表

use dashmap::DashMap;
use std::thread;

fn dashmap_example() {
    let map = DashMap::new();
    
    let mut handles = Vec::new();
    
    for i in 0..10 {
        let handle = thread::spawn(move || {
            map.insert(i, i * 2);
        });
        handles.push(handle);
    }
    
    for handle in handles {
        handle.join().unwrap();
    }
    
    for pair in map.iter() {
        println!("{}: {}", pair.key(), pair.value());
    }
}

六、异步并发

6.1 使用tokio进行异步编程

use tokio;

#[tokio::main]
async fn async_concurrent() {
    let task1 = tokio::spawn(async {
        println!("Task 1 started");
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
        println!("Task 1 completed");
        1
    });
    
    let task2 = tokio::spawn(async {
        println!("Task 2 started");
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        println!("Task 2 completed");
        2
    });
    
    let (result1, result2) = tokio::join!(task1, task2);
    println!("Results: {}, {}", result1.unwrap(), result2.unwrap());
}

6.2 异步安全

use tokio::sync::Mutex;
use std::sync::Arc;

async fn async_mutex() {
    let counter = Arc::new(Mutex::new(0));
    
    let mut handles = Vec::new();
    
    for _ in 0..10 {
        let counter = Arc::clone(&counter);
        let handle = tokio::spawn(async move {
            let mut num = counter.lock().await;
            *num += 1;
        });
        handles.push(handle);
    }
    
    for handle in handles {
        handle.await.unwrap();
    }
    
    println!("Counter: {}", *counter.lock().await);
}