为什么被屏蔽的tokio工人仍然可以并发工作？

Question

当我跑步时

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 循环上工作，因为它永远不会屈服。唯一的工人应该没有机会从事接收工作。因此，第一次发送后通道应该已满。 事实证明我错了。我错过了什么？

Answer 1

#[tokio::main]

函数中的代码实际上并不在工作线程上运行。因此，生成的任务被发送到唯一的工作线程，而

for

循环在程序的主线程上执行。

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

第一种方法是将

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

将失败并显示“没有可用容量”。