在Tokio中向Mutex添加值时出现所有权转移错误的解决办法

我创建了一个共享向量实例：

let shared_vec: Arc<Mutex<Vec<user::User>>> = Arc::new(Mutex::new(Vec::new()));

使用的是tokio::sync::Mutex类型。在tokio::spawn的异步任务中尝试获取锁并向列表添加值时，出现错误：

value moved here, in previous iteration of loop

错误指向tokio::spawn(async move {...}的结尾处。完整代码如下：

use tokio::{
    io::{AsyncBufReadExt, AsyncWriteExt, BufReader},
    net::TcpListener,
    sync::broadcast,
};

use serde_json::json;
use serde_json::Value;
use tokio::sync::Mutex;
use std::sync::Arc;
mod user;

#[tokio::main]
async fn main() {
    let listener = TcpListener::bind("localhost:8080").await.unwrap();
    let (tx, _rx) = broadcast::channel(10);


    let shared_vec: Arc<Mutex<Vec<user::User>>> = Arc::new(Mutex::new(Vec::new()));

    loop {
        let (mut socket, addr) = listener.accept().await.unwrap();
        
        let tx = tx.clone();
        let mut rx = tx.subscribe();

        tokio::spawn(async move {
            let (reader, mut writer) = socket.split();

            let mut reader = BufReader::new(reader);
            let mut line = String::new();

            loop {
                tokio::select! {
                    result = reader.read_line(&mut line) => {
                        if result.unwrap() == 0 {
                            break;
                        }
                        tx.send((line.clone(), addr)).unwrap();
                        line.clear();
                    }
                    result = rx.recv() => {
                        let (msg, other_addr) = result.unwrap();

                        let json_msg: Value = json!(msg);
                        
                        let mut vec = shared_vec.lock().await;

                        let mut indices_to_update = Vec::new();

                        for (index, usr) in vec.iter().enumerate() {
                            if usr.get_ip() != &addr {
                                indices_to_update.push(index);
                            }
                        }

                        for _index in indices_to_update {
                            let user = user::User::new(json_msg["username"].to_string(), addr);
                            vec.push(user);
                        }

                        if addr == other_addr {
                            writer.write_all(msg.as_bytes()).await.unwrap();
                        }
                    }
                }
            }
        });
    }
}

解决方法 #

错误原因 #

async move会捕获并移动环境中的变量到异步任务中。第一次循环时，shared_vec这个Arc实例被移入任务，原变量失去所有权，后续循环迭代再尝试使用它就会触发移动错误。

修复步骤 #

每次循环迭代时，克隆shared_vec的Arc引用（Arc的clone是轻量操作，仅增加引用计数，不复制底层数据），将克隆后的实例移入异步任务，避免原变量被耗尽。

修改后的关键代码 #

在循环内添加Arc克隆：

loop {
    let (mut socket, addr) = listener.accept().await.unwrap();
    
    let tx = tx.clone();
    let mut rx = tx.subscribe();
    // 克隆Arc，生成新的共享引用
    let shared_vec = shared_vec.clone();

    tokio::spawn(async move {
        // 任务内部使用克隆后的shared_vec
        let mut vec = shared_vec.lock().await;
        // ... 原有逻辑保持不变
    });
}

修改后的完整代码 #

use tokio::{
    io::{AsyncBufReadExt, AsyncWriteExt, BufReader},
    net::TcpListener,
    sync::broadcast,
};

use serde_json::json;
use serde_json::Value;
use tokio::sync::Mutex;
use std::sync::Arc;
mod user;

#[tokio::main]
async fn main() {
    let listener = TcpListener::bind("localhost:8080").await.unwrap();
    let (tx, _rx) = broadcast::channel(10);


    let shared_vec: Arc<Mutex<Vec<user::User>>> = Arc::new(Mutex::new(Vec::new()));

    loop {
        let (mut socket, addr) = listener.accept().await.unwrap();
        
        let tx = tx.clone();
        let mut rx = tx.subscribe();
        // 克隆Arc引用
        let shared_vec = shared_vec.clone();

        tokio::spawn(async move {
            let (reader, mut writer) = socket.split();

            let mut reader = BufReader::new(reader);
            let mut line = String::new();

            loop {
                tokio::select! {
                    result = reader.read_line(&mut line) => {
                        if result.unwrap() == 0 {
                            break;
                        }
                        tx.send((line.clone(), addr)).unwrap();
                        line.clear();
                    }
                    result = rx.recv() => {
                        let (msg, other_addr) = result.unwrap();

                        let json_msg: Value = json!(msg);
                        
                        let mut vec = shared_vec.lock().await;

                        let mut indices_to_update = Vec::new();

                        for (index, usr) in vec.iter().enumerate() {
                            if usr.get_ip() != &addr {
                                indices_to_update.push(index);
                            }
                        }

                        for _index in indices_to_update {
                            let user = user::User::new(json_msg["username"].to_string(), addr);
                            vec.push(user);
                        }

                        if addr == other_addr {
                            writer.write_all(msg.as_bytes()).await.unwrap();
                        }
                    }
                }
            }
        });
    }
}

注意事项 #

• Arc的clone操作非常轻量，不会带来性能开销；
• 异步任务中持有锁的时间尽量短，避免长时间阻塞其他任务获取锁，影响并发效率。

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