共享状态并发 - 元素码农

发布时间: 2025-03-22 08:50

↑

# 共享状态并发

Rust的并发模型不仅支持消息传递，还提供了共享状态并发的机制。本文将详细介绍如何在Rust中安全地共享和修改状态。

## 互斥锁（Mutex）

### 1. 基本用法

```rust
use std::sync::Mutex;

fn main() {
    let counter = Mutex::new(0);
    
    {
        let mut num = counter.lock().unwrap();
        *num += 1;
    } // 锁在这里自动释放
    
    println!("Result: {}", *counter.lock().unwrap());
}
```

### 2. 线程间共享

```rust
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 counter = Arc::clone(&counter);
        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());
}
```

## 读写锁（RwLock）

### 1. 基本概念

```rust
use std::sync::RwLock;

fn main() {
    let data = RwLock::new(5);

// 多个读取者
    {
        let r1 = data.read().unwrap();
        let r2 = data.read().unwrap();
        println!("读取值: {} {}", *r1, *r2);
    }

// 单个写入者
    {
        let mut w = data.write().unwrap();
        *w += 1;
    }
}
```

### 2. 并发读写

```rust
use std::sync::{Arc, RwLock};
use std::thread;

fn main() {
    let data = Arc::new(RwLock::new(vec![1, 2, 3, 4]));
    let mut handles = vec![];

// 读取线程
    for _ in 0..3 {
        let data_ref = Arc::clone(&data);
        handles.push(thread::spawn(move || {
            let values = data_ref.read().unwrap();
            println!("读取数据: {:?}", *values);
        }));
    }

// 写入线程
    let data_ref = Arc::clone(&data);
    handles.push(thread::spawn(move || {
        let mut values = data_ref.write().unwrap();
        values.push(5);
    }));

for handle in handles {
        handle.join().unwrap();
    }
}
```

## 原子类型

### 1. 基本操作

```rust
use std::sync::atomic::{AtomicI32, Ordering};

fn main() {
    let atomic_num = AtomicI32::new(0);

// 原子加法
    atomic_num.fetch_add(1, Ordering::SeqCst);
    
    // 原子存储
    atomic_num.store(10, Ordering::SeqCst);
    
    // 原子加载
    let value = atomic_num.load(Ordering::SeqCst);
    println!("当前值: {}", value);
}
```

### 2. 多线程计数器

```rust
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![];

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));
}
```

## 条件变量（Condvar）

### 1. 基本用法

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

fn main() {
    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();
        println!("改变状态...");
        *started = true;
        cvar.notify_one();
    });

let (lock, cvar) = &*pair;
    let mut started = lock.lock().unwrap();
    while !*started {
        started = cvar.wait(started).unwrap();
    }
    println!("状态已改变!");
}
```

### 2. 生产者-消费者模式

```rust
use std::sync::{Arc, Mutex, Condvar};
use std::thread;
use std::collections::VecDeque;

fn main() {
    let queue = Arc::new((Mutex::new(VecDeque::new()), Condvar::new()));
    let queue_producer = Arc::clone(&queue);
    let queue_consumer = Arc::clone(&queue);

// 生产者线程
    let producer = thread::spawn(move || {
        let (queue, cvar) = &*queue_producer;
        for i in 0..5 {
            let mut queue = queue.lock().unwrap();
            queue.push_back(i);
            cvar.notify_one();
            println!("生产: {}", i);
            thread::sleep(std::time::Duration::from_millis(100));
        }
    });

// 消费者线程
    let consumer = thread::spawn(move || {
        let (queue, cvar) = &*queue_consumer;
        let mut total = 0;
        while total < 5 {
            let mut queue = queue.lock().unwrap();
            while queue.is_empty() {
                queue = cvar.wait(queue).unwrap();
            }
            if let Some(val) = queue.pop_front() {
                println!("消费: {}", val);
                total += 1;
            }
        }
    });

producer.join().unwrap();
    consumer.join().unwrap();
}
```

## 最佳实践

### 1. 选择合适的同步原语

- 使用`Mutex`用于简单的互斥访问
- 使用`RwLock`当读操作显著多于写操作时
- 使用原子类型处理简单的计数器或标志
- 使用`Condvar`实现线程间的条件同步

### 2. 避免死锁

```rust
use std::sync::{Arc, Mutex};
use std::thread;

// 正确的锁获取顺序
fn transfer(from: &Mutex<u64>, to: &Mutex<u64>, amount: u64) {
    // 始终按照固定顺序获取锁
    let mut from_balance = from.lock().unwrap();
    let mut to_balance = to.lock().unwrap();
    
    *from_balance -= amount;
    *to_balance += amount;
}
```

### 3. 错误处理

```rust
use std::sync::{Arc, Mutex};

fn process_data(data: Arc<Mutex<Vec<i32>>>) -> Result<(), String> {
    let mut data = data.lock().map_err(|_| "锁已被污染")?;
    if data.is_empty() {
        return Err("数据为空".to_string());
    }
    data.push(42);
    Ok(())
}
```

## 注意事项

1. **性能考虑**：
   - 锁的粒度要适中，既不要太粗也不要太细
   - 持有锁的时间应尽可能短
   - 考虑使用无锁数据结构

2. **安全性**：
   - 避免在持有锁时调用未知代码
   - 正确处理panic情况
   - 避免循环依赖的锁

3. **可维护性**：
   - 清晰地文档化同步策略
   - 使用RAII模式管理资源
   - 保持代码模块化

## 总结

Rust的共享状态并发模型提供了丰富的同步原语，可以安全且高效地处理并发场景。通过合理使用这些工具，我们可以：

1. 安全地共享和修改状态
2. 实现高效的并发控制
3. 避免常见的并发问题

在实际应用中，应根据具体场景选择合适的同步机制，并始终遵循Rust的安全性原则。