如何在Rust(Tokio异步运行时)中实现代码串行执行避免竞态？

问题根因 #

你这问题说白了就是锁用错了：

• std::sync::Mutex是同步锁，在Tokio异步环境里用纯纯找死——它会直接阻塞整个线程池，而且要是锁里包的类型没实现Send，编译直接报错，完全不适合异步场景。
• 换了tokio::sync::Mutex加Arc还出问题，大概率是你没把整个execute的操作链（查余额→算新值→写事务）都放在锁的保护范围内，中间把锁放了，导致并发请求插进来乱改。

正确实现步骤 #

1. 把共享状态包对 #

先把要保护的资源（比如内存里的余额哈希表、数据库连接池）用Arc<tokio::sync::Mutex<T>>包好，而且要保证所有请求用的是同一个实例——要么放全局，要么通过Warp的with方法注入到路由里。

举个例子：

// 定义你的应用全局状态
struct AppState {
    db_pool: sqlx::PgPool,
    // 如果余额存在内存里，就用Mutex包HashMap
    user_balances: tokio::sync::Mutex<std::collections::HashMap<u64, i64>>,
}

// 初始化状态，用Arc包起来让所有请求共享
let app_state = Arc::new(AppState {
    db_pool: pg_pool,
    user_balances: tokio::sync::Mutex::new(std::collections::HashMap::new()),
});

2. 全程攥住锁不松手 #

核心就是：在execute函数开头就把锁拿了，直到整个操作全做完再放（Rust会自动在锁的作用域结束时释放，不用手动管）。

❌ 错误示范（就是你现在出问题的原因）：

async fn execute(state: Arc<AppState>, user_id: u64) -> Result<(), Error> {
    // 查余额：拿锁→读→放锁
    let balance = {
        let balances = state.user_balances.lock().await;
        balances.get(&user_id).copied().unwrap_or(0)
    };
    // 算新余额：没锁，这时候别的请求可能已经改了余额
    let new_balance = balance + 50;
    // 更新：再拿锁→写→放锁
    {
        let mut balances = state.user_balances.lock().await;
        balances.insert(user_id, new_balance);
    }
    Ok(())
}

✅ 正确写法：

async fn execute(state: Arc<AppState>, user_id: u64) -> Result<(), Error> {
    // 一次拿锁，攥到操作全完
    let mut balances = state.user_balances.lock().await;
    
    // 查余额（锁内操作）
    let balance = balances.get(&user_id).copied().unwrap_or(0);
    // 算新余额（锁内，保证基于最新值）
    let new_balance = balance + 50;
    // 写事务/更新内存（锁内完成，全程无空隙）
    sqlx::query!("UPDATE users SET balance = $1 WHERE id = $2", new_balance, user_id)
        .execute(&state.db_pool)
        .await?;
    balances.insert(user_id, new_balance);
    
    Ok(())
}

3. 数据库场景的额外提醒 #

如果余额存在数据库里，光靠应用层的锁还不够，最好结合数据库事务+行级锁，比如用SELECT ... FOR UPDATE把目标用户的行锁住，防止别的事务瞎改：

async fn execute(state: Arc<AppState>, user_id: u64) -> Result<(), Error> {
    // 开数据库事务
    let mut tx = state.db_pool.begin().await?;
    
    // 用SELECT ... FOR UPDATE锁定用户行，别的事务碰不了
    let row = sqlx::query!("SELECT balance FROM users WHERE id = $1 FOR UPDATE", user_id)
        .fetch_one(&mut tx)
        .await?;
    let balance = row.balance;
    let new_balance = balance + 50;
    
    // 更新余额
    sqlx::query!("UPDATE users SET balance = $1 WHERE id = $2", new_balance, user_id)
        .execute(&mut tx)
        .await?;
    
    // 提交事务
    tx.commit().await?;
    Ok(())
}

这种情况下，就算应用层没锁，数据库的行级锁也能保证操作是原子的，不会出竞态。

验证方式 #

用k6测的时候，专门怼同一个用户ID发并发请求，看最终余额是不是初始值加100。如果是数据库场景，还能查一下事务日志，确认没有并发修改的情况。

内容的提问来源于stack exchange，提问作者Arthur