Rust中WebSocket值在tokio::spawn后被移动，如何多次使用？

问题场景 #

需要在两个异步任务中同时使用WebSocket实例：一个任务发送消息，另一个任务接收消息，但遇到以下问题：

1. 直接在tokio::spawn中移动ws后，外部循环无法再使用，触发所有权移动报错
2. 将接收循环移到同一任务后，因第一个reader.recv()循环阻塞，导致WebSocket接收逻辑永远无法执行
3. 参考示例使用ws.next()时，编译器提示该方法未定义

初始代码（所有权错误） #

async fn socket(mut ws: WebSocket, state: Users) {
    tokio::spawn(async move { 
       while let Some(msg)  = reader.recv().await{
           println!("message for user: {:?}", msg);
           ws.send(msg).await.unwrap();
        };
    });

    while let Some(msg) = ws.recv().await{
        // 处理接收消息
    }
}

修改后代码（接收不到消息） #

tokio::spawn(async move {
    while let Some(msg)  = reader.recv().await{
        println!("message for user: {:?}", msg);
        ws.send(msg).await.unwrap();
    };
    while let Some(msg) = ws.recv().await{
        // 处理接收消息
    }   
});

解决方案 #

方案1：用Arc<Mutex<WebSocket>>实现共享所有权 #

通过Arc实现多任务共享所有权，Mutex保证同一时间只有一个任务操作WebSocket，避免并发冲突：

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

async fn socket(ws: WebSocket, state: Users) {
    // 将WebSocket包装为线程安全的共享对象
    let ws = Arc::new(Mutex::new(ws));
    // 克隆Arc给子任务
    let ws_clone = Arc::clone(&ws);

    tokio::spawn(async move {
        while let Some(msg) = reader.recv().await {
            println!("message for user: {:?}", msg);
            // 异步获取锁并发送消息
            ws_clone.lock().await.send(msg).await.unwrap();
        }
    });

    // 获取锁并执行接收循环
    let mut ws = ws.lock().await;
    while let Some(msg) = ws.recv().await {
        // 处理接收到的消息
    }
}

• Arc提供原子引用计数，支持多任务共享同一对象
• tokio::sync::Mutex是异步锁，lock().await会异步等待，不会阻塞线程

方案2：拆分WebSocket读写流（推荐，若库支持） #

多数现代WebSocket库（如tokio-tungstenite）支持拆分读写流为独立的Sink（写端）和Stream（读端），两者可分别移动到不同任务，无需锁：

use tokio_tungstenite::WebSocketStream;
use futures::{SinkExt, StreamExt};

async fn socket(ws: WebSocketStream<...>, state: Users) {
    // 拆分WebSocket为独立的写端和读端
    let (mut write, mut read) = ws.split();

    tokio::spawn(async move {
        while let Some(msg) = reader.recv().await {
            println!("message for user: {:?}", msg);
            write.send(msg).await.unwrap();
        }
    });

    // 读端独立处理接收逻辑
    while let Some(msg) = read.next().await {
        // 处理接收到的消息
    }
}

• 拆分后读写操作互不阻塞，性能更优
• next()方法需要引入futures::StreamExt trait，若你的库实现了Stream即可使用

问题原因说明 #

1. 所有权移动错误：Rust的所有权规则不允许同一值被多个任务拥有，async move会将ws移动到子任务，外部无法再访问
2. 串行循环阻塞：将两个循环放在同一任务时，第一个reader.recv()循环会一直运行到reader关闭，后续的ws.recv()永远不会执行，必须让两个逻辑并发运行
3. next()方法未定义：该方法属于Stream trait，需要确保WebSocket类型实现了Stream，并引入futures::StreamExt扩展 trait

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