主页 / coding / 问题 / 79150081
Accepted
yushizhao
Asked: 2024-11-02 14:23:56 +0800 CST

为什么被屏蔽的tokio worker还能同时工作？

  • 772

当我跑步时

use tokio::sync::mpsc;

#[tokio::main(flavor = "multi_thread", worker_threads = 1)]
async fn main() {
    let (tx, mut rx) = mpsc::channel(1);

    tokio::spawn(async move {
        while let Some(i) = rx.recv().await {
            println!("got = {}", i);
        } 
    });

    for i in 0..5 {

        // busy calculation
        std::thread::sleep(std::time::Duration::from_millis(10));

        match tx.try_send(i) {
            Ok(_) => {
                println!("sent = {}", i);
            },
            Err(err) => {
                println!("{}", err);
            }
        };

    };
 
}

我得到了

sent = 0
got = 0
sent = 1
got = 1
sent = 2
got = 2
sent = 3
got = 3
sent = 4

据我了解，唯一的工作者正在处理 for 循环，因为它从不产生结果。唯一的工作者应该没有机会处理接收。因此，通道在第一次发送后应该是满的。结果我错了。我遗漏了什么？

asynchronous

1 个回答

  1. Best Answer

    函数中的代码#[tokio::main]实际上并不在工作线程上运行。因此，生成的任务被发送到唯一的工作线程，而循环则for在程序的主线程上执行。

    在底层，tokio::main将函数体提升为一个async块，构建一个运行时，然后将生成的异步块的未来传递给Runtime::block_on。根据此方法的文档：

    请注意，此函数所需的 Future 不会作为 Worker 运行。预期是此处的 Future 会生成其他任务。等待此处提供的未来中的其他 Future 的执行速度不会像作为 Worker 生成的 Future 那样快。

    您可以通过两种方式实现您的期望。

    第一种方法是将函数主体提升main到新任务中：

    use tokio::sync::mpsc;

#[tokio::main(flavor = "multi_thread", worker_threads = 1)]
async fn main() {
    tokio::spawn(async move {
        let (tx, mut rx) = mpsc::channel(1);

        tokio::spawn(async move {
            while let Some(i) = rx.recv().await {
                println!("got = {}", i);
            }
        });

        for i in 0..5 {
            // busy calculation
            std::thread::sleep(std::time::Duration::from_millis(10));

            match tx.try_send(i) {
                Ok(_) => {
                    println!("sent = {}", i);
                }
                Err(err) => {
                    println!("{}", err);
                }
            };
        }
    })
    .await
    .unwrap();
}

    第二种是使用“当前线程”运行时，而不是多线程运行时。这将在主线程上运行所有任务。

    use tokio::sync::mpsc;

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let (tx, mut rx) = mpsc::channel(1);

    tokio::spawn(async move {
        while let Some(i) = rx.recv().await {
            println!("got = {}", i);
        }
    });

    for i in 0..5 {
        // busy calculation
        std::thread::sleep(std::time::Duration::from_millis(10));

        match tx.try_send(i) {
            Ok(_) => {
                println!("sent = {}", i);
            }
            Err(err) => {
                println!("{}", err);
            }
        };
    }
}

    这两者都会显示“已发送 = 0”，然后try_send会因“无可用容量”而失败。

    • 2

相关问题