铁锈托基奥“未来不能安全地在线程之间发送"，如何使用watch::channel来共享Vec<u8>数据？

Question

该项目正在实现一个TcpListener服务。它监视pc事件并向TCP客户端广播。作为需求(一个生产者和许多消费者)。但是，代码会引发“未来无法安全地在线程之间发送”错误：

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和其他复杂类型将引发异常。

我在谷歌上搜索，但还是很困惑。谢谢你的帮助。

Answer 1

我认为，因为Ref<_,T>本身并没有实现Send，所以需要将这种类型转换为Send。我认为唯一明智的方法是先克隆它，然后再写出来。似乎很管用。

此外，您还需要使用borrow_and_update将该值标记为“查看”。这是为了确保您不会重复地向stream写入相同的值。

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 coba = rx.borrow_and_update().clone();
            stream.write(&coba).await;
        }
    });

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

    Ok(())
}