advance-practice/channels

[giscus[bot]]

advance-practice/channels

https://course.rs/advance-practice/channels.html

[einQimiaozi]

您好，我想问一下，tokio是不是无法完成golang中使用channel控制并发实现事件驱动并处理消息的功能呢？

我希望通过一个loop不断接收client请求，为每一个client连接的tcpstream分配spawn用于处理，使用一个一对多的channel，将每个client连接的请求通过channel发送

使用一个独立的spawn监听channel，一旦有发送事件到来就接收消息然后处理并将结果回写给client
但我发现channel中的接收通道无法在一个loop中进行async move给监听spawn，原因是因为这样会造成所有权转移，当第二轮loop后，接收通道的所有权已经转移，所以无法通过编译

所以我尝试对tcpstream进行读写分离，通过arc将writehalf在spawn之间进行共享，这样只需要在loop外单独使用一个spawn监听接收通道即可，接收通道的所有权只需要转移给这一个spawn，没有问题

具体的做法是将client请求和该client对应的tcpstream中的writehalf封装成一个message通过channel发送
但是我发现我用于监听的channel在获取message后，只能拿到writehalf的引用，这样根本无法调用writehalf的write相关方法，无法进行回写

目前我也想不到其他的方法能够实现该功能了

我想请教一下是tokio确实无法完成这样的功能，还是我的使用方法不对？

[ghost]

下面这段代码功能是监听udp socket收到的消息, 每次收到的请求使用单独的task去处理，处理完成后再回写到客户端。
使用Arc共享socket，使用两个mspc::channel进行通信。

use tokio::{net::UdpSocket, sync::mpsc, time::{sleep, Duration}};
use std::{io::Result, sync::Arc, net::SocketAddr};

async fn process(data: &[u8]) -> Result<Vec<u8>> {
    let res = Vec::<u8>::new();
    println!("start process, len:{}, sleep 2s", data.len());
    sleep(Duration::from_secs(2)).await;
    println!("finish process, len:{}", data.len());
    Ok(res)
}


async fn start_server(port: u16) -> Result<()> {
    
    let server_addr = format!("0.0.0.0:{port}").parse::<SocketAddr>().unwrap();
    let sock = UdpSocket::bind(server_addr).await.unwrap();
    let reader = Arc::new(sock);
    let sender = reader.clone();

    let (tx, mut rx) = mpsc::channel::<(Vec<u8>, SocketAddr)>(1_000);
    let (tx2, mut rx2) = mpsc::channel::<(Vec<u8>, SocketAddr)>(1_000);

    tokio::spawn(async move { // process data
        while let Some((bytes, addr)) = rx.recv().await {
            let tx3 = tx2.clone();
            tokio::spawn(async move {
                let data = process(&bytes).await.unwrap();
                tx3.send((data, addr)).await.unwrap();
            });
        }
    });

    tokio::spawn(async move { // send data
        while let Some((bytes, addr)) = rx2.recv().await {
            let len = sender.send_to(&bytes, &addr).await.unwrap();
            println!("send to data: {:?}", len);
        }
    });

    let mut buf = [0; 512];
    loop { // receive data
        let (len, addr) = reader.recv_from(&mut buf).await?;
        tx.send((buf[..len].to_vec(), addr)).await.unwrap();
    }
}



#[tokio::main]
async fn main() -> Result<()> {
    start_server(9999).await
}

[einQimiaozi]

感谢！！！我这边也找到了一些新方案，似乎封装一个message结构，在channel中传递message，message中除了数据之外再带一个spawn和主线程单独通信的oneshot信道的方案也可以，这样还不需要共享socket，貌似更安全一些

[zqlpaopao]

不太明白为啥啊要用两个channel
1、tokio的thread是一般是核心数，也即是固定的，直接用多播多接收 会远远好于两个channel(多播单收)的模式，这样数据多了很多传输的过程

[pzxy]

这代码有点乱呀，东一块，西一块的。

[ghost]

确实有点乱, 当时刚好看到这节内容顺手试了一下。
使用async_channel(支持多消费,多生产者,消息只被接收一次)更为简单, 且无需共享socket.

• cargo.toml:

[dependencies]
tokio = {version="1", features=["full"]}
async-channel = {version = "^1.6.1"}
anyhow = {version="^1.0.58"}
bytes = {version="^1.1.0"}

• main.rs

use std::{net::{SocketAddr}, sync::Arc, time::Duration};
use bytes::{Bytes, BytesMut};
use anyhow::{Result};
use async_channel::{bounded, Receiver, Sender};
use tokio::{signal, net::UdpSocket, time::sleep};


/// worker 线程
async fn worker(worker_name: String, req_receiver: Arc<Receiver<(SocketAddr, Bytes)>>, res_sender: Arc<Sender<(SocketAddr, Bytes)>>) -> Result<()>{
    println!("{} running...", worker_name);
    loop {
        tokio::select! {
            res = req_receiver.recv() => {
                let (src, msg) = match res {
                    Ok(r) => r,
                    Err(e) => {
                        println!("{}, Unable to read data, error:{}", worker_name, e);
                        return Ok(());
                    }
                };

                // 处理逻辑, 等待1s...
                sleep(Duration::from_secs(1)).await;
                println!("{}, process the request, src:{:?}", worker_name, src);

                // 结果发送到回复队列
                match res_sender.send((src, msg)).await {
                    Ok(_) => {},
                    Err(e) => println!("{}, Failed to send message to reply queue, error:{}", worker_name, e),
                }
            }
            _ = signal::ctrl_c() => {
                println!("{}, exit...", worker_name);
                req_receiver.close();
                return Ok(());
            }
        }
    }

}

// server线程
async fn server(socket: UdpSocket, req_sender: Arc<Sender<(SocketAddr, Bytes)>>, res_receiver: Arc<Receiver<(SocketAddr, Bytes)>>) -> Result<()> {
    println!("server running...");
    loop {
        let mut buf = BytesMut::with_capacity(1024);
        buf.resize(1024, 0);

        tokio::select! {

            // socket中读取到数据, 发送到请求队列.
            res = socket.recv_from(&mut buf) => {
                let (len, src) = match res {
                    Ok(r) => r,
                    Err(e) => {
                        println!("Unable to read data, error:{}", e);
                        continue;
                    }
                };
                println!("receive socket data, send to request queue. src:{:?}...", src);
                buf.resize(len, 0);
                req_sender.send((src, buf.freeze())).await?;
            }

            // 接收回复队列的数据发送到socket客户端
            Ok((src, msg)) = res_receiver.recv() => {
                println!("receive response queue data, send to socket client:{:?}...", src);
                match socket.send_to(&msg, src).await {
                    Ok(_) => continue,
                    Err(e) => println!("Failed to send data to client({:?}), error:{:?}", src, e),
                }
            }

            _ = signal::ctrl_c() => {
                println!("exit...");
                res_receiver.close();
                return Ok(());
            }

        }
    }
}


#[tokio::main]
async fn main() -> Result<()> {

    let addr = "0.0.0.0:3053".parse::<SocketAddr>()?;
    let socket = UdpSocket::bind(addr).await?;

    // 请求队列
    let (req_sender, req_receiver) = bounded::<(SocketAddr, Bytes)>(1024);
    // 回复队列
    let (res_sender, res_receiver) = bounded::<(SocketAddr, Bytes)>(1024);

    let req_sender = Arc::new(req_sender);
    let req_receiver = Arc::new(req_receiver);
    
    let res_sender = Arc::new(res_sender);
    let res_receiver = Arc::new(res_receiver);

    for i in 0..4 {
        let req_receiver = req_receiver.clone();
        let res_sender = res_sender.clone();
        // worker 线程
        tokio::spawn(async move {
            let worker_name = format!("worker:<{:?}>", i);
            let _ = worker(worker_name, req_receiver, res_sender).await;
        });
    }

    // server 线程
    server(socket, req_sender, res_receiver).await
}

[yAntPower]

有个疑问，在上一节共享状态中的有一小节 “初始化 HashMap” 的代码如下：
use tokio::net::TcpListener;
use std::collections::HashMap;
use std::sync::{Arc, Mutex};

#[tokio::main]
async fn main() {
let listener = TcpListener::bind("127.0.0.1:6379").await.unwrap();

println!("Listening");

let db = Arc::new(Mutex::new(HashMap::new()));

loop {
    let (socket, _) = listener.accept().await.unwrap();
    // 将 handle 克隆一份
    let db = db.clone();

    println!("Accepted");
    tokio::spawn(async move {
        process(socket, db).await;
    });
}

}

我的疑问是 为什么这里的tokio::spawn() 不用.await 而本节的tokio::spawn代码需要用.await 呢？

[limon]

这里是保证 task 在主线程退出前执行完。上一节在 loop 里所以不需要。

[cctyl]

其实提交了就会开始执行，不用.await也可以执行。 用.await只是防止主线程先退出，相当于原本std::thread 里面的join

[xulanwen]

resp_rx 看着是个Receiver类型，为啥可以调用.await呢

[way2ex]

实现了 Future 特征

[Oops-w]

hello 您好，我有个问题想请教一下。
在文中有提到 在 main 函数的末尾，我们让 3 个任务，按照需要的顺序开始运行这么一句话。
但我在实际执行的时候发现，执行的顺序与这三个任务是否使用.await是无关的，随后我也有在tokio的spawn函数中寻找相关信息，库中表明You do not have to .await the returned JoinHandle to make the provided future start execution. It will start running in the background immediately when spawn is called.。那么实际上是否tokio::spawn()与thread::spwan()是相似的呢？实际上两个spawn()都是创建即开始运行，而这三个任务的.await实际上只是为了能够等待其他线程执行完之后再结束main线程并且它是不能够指定执行顺序的？

[MorseWayne]

我直接去询问了一下ChatGPT，得到如下解释
"在 Rust 中，tokio::spawn 确实会生成新的异步任务。然而，这些任务的生命周期是与创建它们的 Tokio 运行时绑定的。也就是说，如果 Tokio 运行时结束了，那么它创建的所有任务也会被结束。

在你的例子中，Tokio 运行时是由 #[tokio::main] 注解创建的。这个注解会创建一个 Tokio 运行时，并在这个运行时中运行 main 函数。当 main 函数返回时，Tokio 运行时就会结束，同时它创建的所有任务也会被结束。

所以，即使 t1 和 t2 是新的异步任务，但是如果你在 main 函数中没有等待它们完成（即没有调用 .await），那么 main 函数可能会在这些任务完成之前就返回。这将导致 Tokio 运行时结束，从而提前终止这些任务。

因此，为了确保所有的异步任务都能够正常完成，我们需要在 main 函数中等待它们完成。这就是为什么你需要在 main 函数中调用 t1.await; 和 t2.await; 的原因。"

[woaiqiukui]

把t2.await.unwrap();做了注释依然会去执行t2任务，请问是let tx2 = tx.clone();的问题嘛，把tx2发送器注释了就不会执行了

[bxb100]

没错, 只有 drop 的时候 tx_count 减为 0 的时候才能真正 tx close 掉

见 tokio-1.26.0/src/sync/mpsc/chan.rs:207

[runzhi214]

是的没错，使用while let监听的话，sender没有全部drop的话很容易陷入死锁。

[fatli]

上面提到：“有一点值得注意，往 oneshot 中发送消息时，并没有使用 .await，原因是该发送操作要么直接成功、要么失败，并不需要等待。”
但是看代码示例 ，都用了await呢。请问是指的哪哦！还是我理解 错误了么。

[FlippedOvO]

Command::Set { key, val, resp } => {
    let res = client.set(&key, val).await;
    // 忽略错误
    let _ = resp.send(res);
}

这个resp 是 oneshot 的 Sender

[samlior]

while let Some(cmd) = rx.recv().await {
    match cmd {
        Command::Get { key, resp } => {
            let res = client.get(&key).await;
            // 忽略错误
            let _ = resp.send(res);
        }
        Command::Set { key, val, resp } => {
            let res = client.set(&key, val).await;
            // 忽略错误
            let _ = resp.send(res);
        }
    }
}

这个地方发往redis的请求还是串行的, 怎么做到并行? 在第一个请求得到结果之前就发出第二个请求

[featherL]

我也有这个疑惑，虽然说发送端并发了，但是接受端还是串行的，依然还是串行的调用 client.get，这个例子引入通道感觉很牵强

[TimKong1]

有一点值得注意，往 oneshot 中发送消息时，并没有使用 .await，原因是该发送操作要么直接成功、要么失败，并不需要等待。

当 oneshot 的接受端被 drop 后，继续发送消息会直接返回 Err 错误，它表示接收者已经不感兴趣了。对于我们的场景，接收者不感兴趣是非常合理的操作，并不是一种错误，因此可以直接忽略。

有点不是很理解，上面不是写到Future不await的时候是不会执行的吗 怎么resp.send(res);会执行呢？

[Tuffyyy]

问了AI，说是resp.send是一个普通的函数调用

“在代码中调用 resp.send(res)（假设 resp 是 oneshot 通道的发送端）时，它只是在执行一个普通的函数调用，这个函数内部会立即尝试进行发送消息的操作，不管成功还是失败都会即刻返回对应的结果（成功则返回 Ok，失败返回 Err），它本身并不涉及像异步 Future 那样需要等待外部条件满足或者持续执行一段时间的逻辑，所以不需要使用 .await 来等待其完成，它在调用的当下就已经执行了发送尝试这个动作，并且返回了相应的结果。

也就是说，虽然它返回的结果类型可能实现了 Future 特性（例如 Result<(), SendError> 这种形式相关的类型在 oneshot 里可能涉及到 Future 相关的一些底层实现细节用于兼容异步编程的错误处理等场景），但就其核心的发送消息这个行为本身是即时的，不是那种需要等待的异步操作，所以不需要使用 .await 去等待其完成。”