Rust 异步并发安全与内存管理实战指南

死锁与活锁

死锁通常源于锁获取顺序不一致。活锁则是任务不断重试却因资源冲突无法推进。解决死锁的关键是统一锁获取顺序；活锁则需要引入随机退避或超时机制。

use std::sync::Mutex;
use tokio::spawn;

#[tokio::main]
async fn main() {
    let mutex1 = Mutex::new(1);
    let mutex2 = Mutex::new(2);
    
    let handle1 = spawn(async move {
        let lock1 = mutex1.lock().unwrap();
        // 模拟延迟，让 handle2 有机会拿到 mutex2
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
        let lock2 = mutex2.lock().unwrap(); // 等待 mutex2
    });
    
    let handle2 = spawn(async move {
        let lock2 = mutex2.lock().unwrap();
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
        let lock1 = mutex1.lock().unwrap(); // 等待 mutex1
    });
    
    handle1.await.unwrap();
    handle2.await.unwrap();
}

资源泄漏

任务被取消后未清理资源，或 Arc 引用计数未正确归零，都会导致内存泄漏。务必确保 Drop 实现正确，并在任务结束时检查状态。

解决方案与最佳实践

使用 Arc + Mutex 保护共享状态

对于需要修改的共享数据，Arc<Mutex<T>> 是最常用的组合。注意在异步环境中应使用 tokio::sync::Mutex 而非标准库的 std::sync::Mutex，后者可能导致阻塞整个运行时。

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

#[tokio::main]
async fn main() {
    let data = Arc::new(Mutex::new(0));
    let mut handles = Vec::new();
    
    for _ in 0..10 {
        let data_clone = Arc::clone(&data);
        handles.push(spawn(async move {
            for _ in 0..1000 {
                let mut lock = data_clone.lock().await;
                *lock += 1;
            }
        }));
    }
    
    for handle in handles {
        handle.await.unwrap();
    }
    println!("Data: {}", data.lock().await);
}

读写分离：Arc + RwLock

如果读多写少，RwLock 能显著提升性能。它允许多个读者同时访问，但写者独占。

use std::sync::Arc;
use tokio::sync::RwLock;
use tokio::spawn;

#[tokio::main]
async fn main() {
    let data = Arc::new(RwLock::new(0));
    let mut handles = Vec::new();
    
    // 5 个写任务
    for _ in 0..5 {
        let data_clone = Arc::clone(&data);
        handles.push(spawn(async move {
            for _ in 0..1000 {
                let mut lock = data_clone.write().await;
                *lock += 1;
            }
        }));
    }
    
    // 5 个读任务
    for _ in 0..5 {
        let data_clone = Arc::clone(&data);
        handles.push(spawn(async move {
            for _ in 0..1000 {
                let lock = data_clone.read().await;
                println!("Data: {}", *lock);
            }
        }));
    }
    
    for handle in handles {
        handle.await.unwrap();
    }
}

原子操作与消息传递

对于简单计数器，AtomicI32 比锁更高效。而对于复杂对象，通道（Channel）是首选，通过'不共享内存，只传递消息'来彻底规避竞态条件。

use tokio::sync::mpsc;
use tokio::spawn;

#[tokio::main]
async fn main() {
    let (sender, mut receiver) = mpsc::channel(10);
    let mut handles = Vec::new();
    
    for _ in 0..10 {
        let sender_clone = sender.clone();
        handles.push(spawn(async move {
            for i in 0..1000 {
                sender_clone.send(i).await.unwrap();
            }
        }));
    }
    
    handles.push(spawn(async move {
        let mut sum = 0;
        while let Some(msg) = receiver.recv().await {
            sum += msg;
        }
        println!("Sum: {}", sum);
    }));
    
    drop(sender); // 关闭发送端
    for handle in handles {
        handle.await.unwrap();
    }
}

内存管理优化

避免频繁分配

在高频循环中，预分配缓冲区或使用对象池能减少 GC 压力（虽然 Rust 无 GC，但分配开销依然存在）。BytesMut 适合处理二进制流，但跨任务共享需加锁。

任务栈大小配置

默认栈大小可能不足。使用 Builder 调整线程栈大小，防止递归调用导致的崩溃。

use tokio::runtime::Builder;
use tokio::time::sleep;
use std::time::Duration;

fn main() {
    let runtime = Builder::new_multi_thread()
        .worker_threads(4)
        .thread_stack_size(2 * 1024 * 1024) // 2MB
        .build()
        .unwrap();
    
    runtime.block_on(async {
        let mut handles = Vec::new();
        for _ in 0..10 {
            handles.push(tokio::spawn(async move {
                sleep(Duration::from_millis(100)).await;
            }));
        }
        for handle in handles {
            handle.await.unwrap();
        }
    });
}

实战项目中的优化

在实际业务中，我们常遇到 HTTP 缓存、数据库连接池和 Redis 客户端的并发问题。

HTTP 客户端缓存

使用 Arc<Mutex<HashMap>> 实现线程安全的本地缓存，避免重复请求。

use reqwest::{Client, Response};
use tokio::sync::Mutex;
use std::sync::Arc;
use std::collections::HashMap;

pub struct HttpClient {
    client: Client,
    cache: Arc<Mutex<HashMap<String, String>>>,
}

impl HttpClient {
    pub fn new() -> Self {
        HttpClient {
            client: Client::new(),
            cache: Arc::new(Mutex::new(HashMap::new())),
        }
    }
    
    pub async fn get(&self, url: &str) -> Result<String, Box<dyn std::error::Error>> {
        let mut cache = self.cache.lock().await;
        if let Some(cached) = cache.get(url) {
            return Ok(cached.clone());
        }
        let response = self.client.get(url).send().await?;
        let body = response.text().await?;
        cache.insert(url.to_string(), body.clone());
        Ok(body)
    }
}

数据库连接池

使用 sqlx 或 deadpool 管理连接，确保连接数上限可控，避免耗尽系统资源。

use sqlx::PgPool;

pub async fn create_pool(config_url: &str) -> Result<PgPool, sqlx::Error> {
    let pool = sqlx::postgres::PgPoolOptions::new()
        .max_connections(10)
        .min_connections(2)
        .connect(config_url)
        .await?;
    Ok(pool)
}

限流控制

在高并发场景下，使用 Semaphore 限制并发度，保护下游服务不被压垮。

use tokio::sync::Semaphore;
use std::sync::Arc;

async fn sync_users(semaphore: Arc<Semaphore>) {
    let permit = semaphore.acquire_owned().await.unwrap();
    // 执行耗时操作
    drop(permit); // 释放信号量
}

总结

Rust 的异步并发安全依赖于对所有权模型的深刻理解。通过合理运用 Arc、Mutex、RwLock 及 mpsc 通道，我们可以有效规避数据竞争和死锁。在内存层面，关注对象复用和任务栈配置同样关键。希望这些实践能帮助你在生产环境中构建更稳健的 Rust 应用。