向tokio::spawn传入Arc<Mutex>时future无法跨线程安全发送报错

错误根因 #

• tokio::spawn 生成的异步任务必须实现 Send 特征：tokio多线程调度器会在.await异步暂停点将任务移动到其他工作线程继续执行，所有跨暂停点持有的值都必须满足Send约束。
• 你使用的std::sync::Mutex是同步场景设计的锁，其对应的锁守卫MutexGuard没有实现Send特征，锁守卫和持有它的线程严格绑定，无法跨线程传递。
• 你的代码中stream.lock().unwrap()得到的锁守卫生命周期覆盖了整个match代码块，块内部存在stream.write().await、stream.read().await这类异步暂停点，相当于持有非Send的锁守卫跨了await调度点，直接违反spawn的Send约束，因此编译失败。
• 额外隐患：即使绕过Send检查，在异步代码中使用阻塞的std::sync::Mutex也会卡死tokio工作线程：等待锁的过程中线程不会让出执行权，会阻塞同线程上其他所有异步任务的调度。

修复方案 #

方案1：快速修复（替换为tokio异步锁） #

将所有跨异步任务共享的std::sync::Mutex替换为tokio::sync::Mutex，它的锁守卫实现了Send特征，且锁等待过程是异步的，不会阻塞工作线程。

注意：该方案下读写任务会竞争同一把锁，同一时间只能执行读或者写，无法利用TcpStream全双工读写的特性，仅适合快速修复编译问题。

修复后的handle_write示例：

// 引入tokio的异步锁，替换原有的std::sync::Mutex
use tokio::sync::Mutex;

async fn handle_write(&mut self) -> JoinHandle<()> {
    let stream = Arc::clone(&self.stream);
    let session = Arc::clone(&self.session);
    let queue = Arc::clone(&self.queue);

    tokio::spawn(async move {
        match stream.lock().await.as_mut() {
            Some(stream) => {
                let packet: Vec<u8> = queue.lock().await.pop_front().unwrap();
                let packet = match session.lock().await.header_crypt.as_mut() {
                    Some(header_crypt) => header_crypt.encrypt(&packet),
                    _ => packet,
                };

                stream.write(&packet).await.unwrap();
                stream.flush().await.unwrap();
            },
            _ => {},
        };
    })
}

handle_read方法按照相同逻辑替换锁、将lock().unwrap()改为lock().await即可解决编译错误。

方案2：最优实践（拆分TcpStream读写半连接，彻底消除锁） #

tokio::net::TcpStream提供了into_split()方法，可以将单个TcpStream拆分为独立的OwnedReadHalf（读半连接）和OwnedWriteHalf（写半连接），两个半连接各自实现Send，可以分别移动到读任务、写任务中独立使用，完全不需要锁包裹共享，既没有锁竞争性能损耗，也从根源上避免了Send相关的编译错误。
实现思路：

• 建立Tcp连接后，直接调用stream.into_split()得到读、写两个独立的半连接
• 写任务独占写半连接所有权，读任务独占读半连接所有权，不需要用Arc<Mutex<>>包裹TcpStream
• 消息队列、会话上下文这类需要跨任务共享的数据，再根据场景选择合适的异步同步原语

核心示例代码：

// 建立连接后直接拆分读写半连接
let (read_half, write_half) = stream.into_split();

// 写任务直接持有write_half所有权，无需加锁
tokio::spawn(async move {
    loop {
        let packet = queue.lock().await.pop_front().unwrap();
        let packet = match session.lock().await.header_crypt.as_mut() {
            Some(crypt) => crypt.encrypt(&packet),
            _ => packet
        };
        write_half.write_all(&packet).await.unwrap();
        write_half.flush().await.unwrap();
    }
});

// 读任务直接持有read_half所有权，无需加锁
tokio::spawn(async move {
    let mut buffer = [0u8; 4096];
    loop {
        let bytes_count = match read_half.read(&mut buffer).await {
            Ok(n) if n > 0 => n,
            _ => break
        };
        let raw_data = match session.lock().await.header_crypt.as_mut() {
            Some(crypt) => crypt.decrypt(&buffer[..bytes_count]),
            _ => buffer[..bytes_count].to_vec()
        };
        queue.lock().await.push_back(raw_data);
    }
});

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