Rust Tokio：使用watch::channel共享Vec<u8>时跨线程安全错误求助

我在实现一个TcpListener服务，用于监控PC事件并向TCP客户端广播数据，符合单生产者多消费者的需求，但编写的代码触发了future cannot be sent between threads safely错误：

use tokio;
use tokio::{io::AsyncWriteExt, net::TcpStream, sync::watch};
#[tokio::main]
async fn main() -> anyhow::Result<()> {
    
    let (tx, mut rx) = watch::channel::<Vec<u8>>(vec![]);

    let mut stream = TcpStream::connect("127.0.0.1:8080").await?;

    tokio::spawn(async move {
        while rx.changed().await.is_ok() {
            stream.write(&*rx.borrow()).await;
        }
    });

    tx.send(vec![1u8, 2u8])?;

    Ok(())
}

运行后报错：

error: "future cannot be sent between threads safely"
   --> src/main.rs:9:18
    |
9   |       tokio::spawn(async move {
    |  __________________^
10  | |         while rx.changed().await.is_ok() {
11  | |             stream.write(&*rx.borrow()).await;
12  | |         }
13  | |     });
    | |_____^ future created by async block is not `Send`
    |
    = help: within `impl Future<Output = ()>`, the trait `Send` is not implemented for `std::sync::RwLockReadGuard<'_, Vec<u8>>`
note: future is not `Send` as this value is used across an await
   --> src/main.rs:11:40
    |
11  |             stream.write(&*rx.borrow()).await;
    |                            ----------- ^^^^^^ await occurs here, with `rx.borrow()` maybe used later
    |                            |
    |                            has type `tokio::sync::watch::Ref<'_, Vec<u8>>` which is not `Send`
note: `rx.borrow()` is later dropped here
   --> src/main.rs:11:46
    |
11  |             stream.write(&*rx.borrow()).await;
    |                                              ^
help: consider moving this into a `let` binding to create a shorter lived borrow
   --> src/main.rs:11:27
    |
11  |             stream.write(&*rx.borrow()).await;
    |                           ^^^^^^^^^^^^
note: required by a bound in `tokio::spawn`
   --> /playground/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-1.22.0/src/task/spawn.rs:163:21
    |
163 |         T: Future + Send + 'static,
    |                     ^^^^ required by this bound in `tokio::spawn`

当我将Vec<u8>改为u8或&str时代码可正常运行，但使用Vec、String等复杂类型就会触发该错误。

原因分析 #

核心问题在于tokio::sync::watch::Ref（即rx.borrow()的返回值）并非Send类型——它持有底层RwLock的读锁，跨线程传递这类锁守卫可能引发死锁风险，因此Rust默认不给它实现Send。

当你直接在stream.write(&*rx.borrow()).await中使用rx.borrow()时，编译器会认为这个锁守卫的生命周期覆盖了await调用：在IO操作挂起期间，守卫仍然被持有。而tokio::spawn要求传入的future必须是Send类型（因为Tokio的线程池可能会在不同线程间调度task），持有非Send值跨await直接导致future不满足Send约束。

对于u8这类Copy类型，编译器会自动优化：解引用Ref后直接复制值，锁守卫在await前就被释放，不会跨线程持有，因此不会触发错误。但Vec、String这类非Copy类型必须依赖锁守卫来维持引用有效性，守卫会被持有到await结束，从而触发错误。

解决方案 #

有两种常用的修复方式：

方式1：克隆数据，脱离锁守卫依赖 #

通过克隆通道中的数据，彻底摆脱对锁守卫的持有，这样future不再包含非Send的锁守卫：

use tokio;
use tokio::{io::AsyncWriteExt, net::TcpStream, sync::watch};
#[tokio::main]
async fn main() -> anyhow::Result<()> {
    
    let (tx, mut rx) = watch::channel::<Vec<u8>>(vec![]);

    let mut stream = TcpStream::connect("127.0.0.1:8080").await?;

    tokio::spawn(async move {
        while rx.changed().await.is_ok() {
            // 克隆数据，锁守卫在此行结束后立即释放
            let data = rx.borrow().clone();
            let _ = stream.write(&data).await;
        }
    });

    tx.send(vec![1u8, 2u8])?;

    Ok(())
}

方式2：提前绑定守卫，缩短其生命周期 #

将锁守卫绑定到局部变量，确保它在await前被释放（编译器会明确守卫的生命周期不覆盖await）：

use tokio;
use tokio::{io::AsyncWriteExt, net::TcpStream, sync::watch};
#[tokio::main]
async fn main() -> anyhow::Result<()> {
    
    let (tx, mut rx) = watch::channel::<Vec<u8>>(vec![]);

    let mut stream = TcpStream::connect("127.0.0.1:8080").await?;

    tokio::spawn(async move {
        while rx.changed().await.is_ok() {
            // 绑定守卫到局部变量
            let guard = rx.borrow();
            // 仅传递引用给write，守卫在当前代码块结束时释放，不跨await
            let _ = stream.write(&*guard).await;
        }
    });

    tx.send(vec![1u8, 2u8])?;

    Ok(())
}

两种方式都能解决Send约束问题，选择哪种取决于你的性能需求：克隆数据会产生内存开销，但逻辑更简单；绑定守卫的方式无额外开销，但需要确保守卫生命周期不跨await。

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