Rust线程间共享数据报错：value moved into closure here, in previous iteration of loop

问题本质 #

E0382错误是因为你在循环迭代中把rx（通道接收端）的所有权转移到了闭包里，导致下一次循环时原变量已失效。Rust的所有权规则不允许重复使用已被转移所有权的变量。

核心解决方案：用Arc共享所有权 #

通过Arc（原子引用计数）包装通道端点，让每个闭包都能持有一份克隆的引用，避免所有权转移。具体步骤如下：

1. 用Arc包裹需要共享的通道端点 #

如果是HTTP服务器要发送消息给WebSocket客户端，应该共享Sender（因为HTTP端需要发消息）；如果是WebSocket端要接收HTTP的消息，共享Receiver。示例中以HTTP发、WebSocket收为例：

use tokio::sync::mpsc;
use std::sync::Arc;

// 创建异步通道
let (tx, rx) = mpsc::channel::<String>(100);
// 用Arc包裹Sender，供HTTP线程共享
let tx_arc = Arc::new(tx);
// 用Arc包裹Receiver，供WebSocket线程使用
let rx_arc = Arc::new(rx);

2. 在HTTP处理闭包中克隆Arc #

每次处理HTTP请求时，克隆一份tx_arc，闭包使用克隆后的实例发送消息：

use axum::{Router, routing::post, Json, State};

let app = Router::new()
    .route("/send", post(|Json(req): Json<String>, tx: State<Arc<mpsc::Sender<String>>>| async move {
        // 发送消息到通道，无需转移所有权
        let _ = tx.send(req).await;
        Ok("Message forwarded to WebSocket client".to_string())
    }))
    .with_state(tx_arc);

3. WebSocket线程中使用克隆的Arc #

WebSocket线程持有rx_arc的克隆，循环接收消息并转发给客户端：

use tokio::net::TcpListener;
use tungstenite::protocol::Message;

let rx_clone = Arc::clone(&rx_arc);
tokio::spawn(async move {
    let listener = TcpListener::bind("127.0.0.1:8081").await.unwrap();
    while let Ok((stream, _)) = listener.accept().await {
        let (mut ws_stream, _) = tungstenite::accept(stream).unwrap();
        let rx = Arc::clone(&rx_clone);
        
        tokio::spawn(async move {
            while let Ok(msg) = rx.recv().await {
                ws_stream.send(Message::Text(msg)).await.unwrap();
            }
        });
    }
});

关键注意事项 #

• 必须使用异步通道（比如tokio::sync::mpsc），因为HTTP和WebSocket都是异步场景，同步通道会阻塞事件循环。
• 不要直接移动原始的tx或rx到闭包，必须通过Arc::clone()创建引用克隆。
• 如果你的场景是WebSocket发消息给HTTP，只需调换tx和rx的共享逻辑即可。

完整可运行示例片段 #

use tokio::sync::mpsc;
use tokio::net::TcpListener;
use axum::{Router, routing::post, Json, State};
use tungstenite::protocol::Message;
use std::sync::Arc;

#[tokio::main]
async fn main() {
    // 初始化异步通道
    let (tx, rx) = mpsc::channel::<String>(100);
    let tx_arc = Arc::new(tx);
    let rx_arc = Arc::new(rx);

    // 启动WebSocket服务
    let rx_clone = Arc::clone(&rx_arc);
    tokio::spawn(async move {
        let listener = TcpListener::bind("127.0.0.1:8081").await.unwrap();
        while let Ok((stream, _)) = listener.accept().await {
            let (mut ws_stream, _) = tungstenite::accept(stream).unwrap();
            let rx = Arc::clone(&rx_clone);
            
            tokio::spawn(async move {
                while let Ok(msg) = rx.recv().await {
                    if let Err(e) = ws_stream.send(Message::Text(msg)).await {
                        eprintln!("Failed to send message: {}", e);
                        break;
                    }
                }
            });
        }
    });

    // 启动HTTP服务
    let app = Router::new()
        .route("/send", post(|Json(req): Json<String>, tx: State<Arc<mpsc::Sender<String>>>| async move {
            match tx.send(req).await {
                Ok(_) => Ok("Message sent successfully".to_string()),
                Err(e) => Err(format!("Failed to send message: {}", e)),
            }
        }))
        .with_state(tx_arc);

    let listener = TcpListener::bind("127.0.0.1:8080").await.unwrap();
    axum::serve(listener, app).await.unwrap();
}

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