使用Tokio无界通道出现值移动错误E0382的解决咨询

我使用Tokio 1.17.0（开启full特性）的unbounded_channel发送SSE消息，由于需要从不同位置发送消息，编写了如下最小复现代码：

use std::sync::{Arc, Mutex};
use tokio::{
    sync::mpsc::{UnboundedReceiver, UnboundedSender},
    task,
};

#[tokio::main]
async fn main() {
    let (tx, rx): (UnboundedSender<String>, UnboundedReceiver<String>) =
        tokio::sync::mpsc::unbounded_channel();
    task::spawn_blocking(move || {
        let shared_tx = Arc::new(Mutex::new(tx));
        shared_tx.lock().unwrap().send("l".to_string());
    });
    tx.send("d".to_string());
}

编译时触发以下错误：

> cargo build
   Compiling rust-learn v0.1.0 (/Users/xiaoqiangjiang/source/reddwarf/backend/rust-learn)
warning: unused variable: `rx`
 --> src/main.rs:9:14
  |
9 |     let (tx, rx): (UnboundedSender<String>, UnboundedReceiver<String>) =
  |              ^^ help: if this is intentional, prefix it with an underscore: `_rx`
  |
  = note: `#[warn(unused_variables)]` on by default

error[E0382]: borrow of moved value: `tx`
  --> src/main.rs:15:5
   |
9  |     let (tx, rx): (UnboundedSender<String>, UnboundedReceiver<String>) =
   |          -- move occurs because `tx` has type `UnboundedSender<String>`, which does not implement the `Copy` trait
10 |         tokio::sync::mpsc::unbounded_channel();
11 |     task::spawn_blocking(move || {
   |                          ------- value moved into closure here
12 |         let shared_tx = Arc::new(Mutex::new(tx));
   |                                             -- variable moved due to use in closure
...
15 |     tx.send("d".to_string());
   |     ^^^^^^^^^^^^^^^^^^^^^^^^ value borrowed here after move

For more information about this error, try `rustc --explain E0382`.
warning: `rust-learn` (bin "rust-learn") generated 1 warning
error: could not compile `rust-learn` due to previous error; 1 warning emitted

我的Cargo.toml配置如下：

[package]
name = "rust-learn"
version = "0.1.0"
edition = "2018"

[dependencies]
tokio = { version = "1.17.0", features = ["full"] }
serde = { version = "1.0.64", features = ["derive"] }
serde_json = "1.0.64"

修复方案 #

问题核心是**UnboundedSender不实现Copy trait**，直接将tx移动进spawn_blocking的闭包后，外部的tx就失去了所有权，无法再使用。正确的处理方式是利用UnboundedSender本身支持克隆且线程安全的特性，用Arc包裹后共享：

use std::sync::Arc;
use tokio::{
    sync::mpsc::{UnboundedReceiver, UnboundedSender},
    task,
};

#[tokio::main]
async fn main() {
    let (tx, _rx): (UnboundedSender<String>, UnboundedReceiver<String>) =
        tokio::sync::mpsc::unbounded_channel();
    // 用Arc包裹Sender，实现多线程共享
    let shared_tx = Arc::new(tx);
    
    // 克隆Arc传递给闭包，避免移动原始实例
    let tx_clone = shared_tx.clone();
    task::spawn_blocking(move || {
        tx_clone.send("l".to_string()).unwrap();
    });
    
    // 直接使用共享的Sender发送消息
    shared_tx.send("d".to_string()).unwrap();
}

关键说明 #

• UnboundedSender本身实现了Send + Sync，是线程安全的，且支持clone方法，不需要额外用Mutex包裹。
• 通过Arc::clone()创建引用计数副本，闭包使用这个副本不会影响原始shared_tx的所有权，外部依然可以正常使用。
• 将rx改为_rx，消除未使用变量的警告。

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