Rust tokio 如何以异步非阻塞方式运行大量任务

tokio 官方给了一个完整的例子：手动构建 runtime ，利用 block_on 来运行多个任务。tokio 的任务是由 之类的函数产生的 类型，而且是个 。

而下面利用 和 await 编写了等价的版本（为了直观对比任务完成的实际顺序和总耗时，我对 sleep 的时间做了一些简化）：
 

usestd::time::Instant; usetokio::time::; #[tokio::main] asyncfnmain() -> std::io::Result {   letnow = Instant::now();   letmuthandles =Vec::with_capacity(10);   foriin..10{     handles.push(tokio::spawn(my_bg_task(i)));   }   // Do something time-consuming while the background tasks execute.   std::thread::sleep(Duration::from_millis(120));   println!("Finished time-consuming task.");   // Wait for all of them to complete.   forhandleinhandles {     handle.await?;   }   println!("总耗时：{} ms", now.elapsed().as_millis());   Ok(()) } asyncfnmy_bg_task(i:u64) {   letmillis =100;   println!("Task {} sleeping for {} ms.", i, millis);   sleep(Duration::from_millis(millis)).await;   println!("Task {} stopping.", i); }

输出结果：

Tasksleepingfor100ms. Task1sleepingfor100ms. Task2sleepingfor100ms. Task3sleepingfor100ms. Task4sleepingfor100ms. Task5sleepingfor100ms. Task6sleepingfor100ms. Task7sleepingfor100ms. Task8sleepingfor100ms. Task9sleepingfor100ms. Task9stopping. Taskstopping. Task1stopping. Task2stopping. Task3stopping. Task4stopping. Task5stopping. Task6stopping. Task7stopping. Task8stopping. Finishedtime-consuming task.

总耗时：120ms
如果把主线程的的 sleep 时间改成 100 ms： 则产生下面的结果：
 

Tasksleepingfor100ms. Task1sleepingfor100ms. Task2sleepingfor100ms. Task3sleepingfor100ms. Task4sleepingfor100ms. Task5sleepingfor100ms. Task6sleepingfor100ms. Task7sleepingfor100ms. Task8sleepingfor100ms. Task9sleepingfor100ms. Finishedtime-consuming task. Task3stopping. Taskstopping. Task1stopping. Task2stopping. Task9stopping. Task4stopping. Task5stopping. Task6stopping. Task7stopping. Task8stopping.

总耗时：103ms
可以看到， 实际是异步非阻塞执行的 ：
异步：因为每个任务不必等待其结果就可以开始下一个任务，即；
// 异步
Tasksleepingfor100ms.
Task1sleepingfor100ms.
...
// 同步
Tasksleepingfor100ms.
Taskstopping.
Task1sleepingfor100ms.
Task1stopping.
...

非阻塞：每个任务之间可以快速切换，不必等待其他任务完成才切换，这个例子表现在：
任务 0-9 以乱序方式 stop
与 的打印顺序只与任务各自的运行 (sleep) 时间有关，与源代码的声明执行顺序无关。只有任务之间快速切换才能做到这一点。回顾官网的例子：10 个任务的 sleep 时间线性递减 （），从 6 个任务开始小于主线程 sleep 任务的时间（750 ms），而等待 10 个任务执行的语句 显然位于 之后，所以任务之间非阻塞执行的话，打印结果为 sleep 时间越短的任务先完成，时间越长的任务后完成，总耗时为任务中的最长耗时：
 

Tasksleepingfor1000ms. Task1sleepingfor950ms. Task2sleepingfor900ms. Task3sleepingfor850ms. Task4sleepingfor800ms. Task5sleepingfor750ms. Task6sleepingfor700ms. Task7sleepingfor650ms. Task8sleepingfor600ms. Task9sleepingfor550ms. Task9stopping. Task8stopping. Task7stopping. Task6stopping. Finished time-consuming task. Task5stopping. Task4stopping. Task3stopping. Task2stopping. Task1stopping. Taskstopping.

总耗时：1001ms// 非常完美
一般情况下，对于 async block/fn 你至少有以下一些做法：
对 async block/fn 调用 来等待结果；
对可列举的少数 Future 调用 或者 来同时等待多个结果 或者 等待多个分支的第一个结果；
对大量 Future 调用 join 或者 select 一类支持传入 Vec / iter 参数类型的函数，比如这个例子中的 部分就可以改写成 ；
把 async block/fn 变成任务，然后调用 （等价地，对任务 await）来执行许多任务。
容易犯的错误是，希望异步非阻塞时，对所有 async block/fn 进行了 await，而没有进行任务化处理（即 把 Future 通过 spwan 函数转化成任务）：

usestd::time::Instant; usetokio::time::; #[tokio::main] asyncfnmain() { letnow = Instant::now(); letmuthandles =Vec::with_capacity(10); foriin..10{ handles.push(my_bg_task(i));// 没有把 Future 变成任务 } std::thread::sleep(Duration::from_millis(120)); println!("Finished time-consuming task."); forhandleinhandles { handle.await;// 而且每个 handle 必须执行完才能执行下一个 handle } println!("总耗时：{} ms", now.elapsed().as_millis()); } asyncfnmy_bg_task(i:u64) { letmillis =100; println!("Task {} sleeping for {} ms.", i, millis); sleep(Duration::from_millis(millis)).await; println!("Task {} stopping.", i); }

运行结果：同步阻塞
 

Finishedtime-consuming task. Tasksleepingfor100ms. Taskstopping. Task1sleepingfor100ms. Task1stopping. Task2sleepingfor100ms. Task2stopping. Task3sleepingfor100ms. Task3stopping. Task4sleepingfor100ms. Task4stopping. Task5sleepingfor100ms. Task5stopping. Task6sleepingfor100ms. Task6stopping. Task7sleepingfor100ms. Task7stopping. Task8sleepingfor100ms. Task8stopping. Task9sleepingfor100ms. Task9stopping.

总耗时：1130ms
或者像这样：
 

usestd::time::Instant; usetokio::time::; #[tokio::main] asyncfnmain() { letnow = Instant::now(); letmuthandles =Vec::with_capacity(10); foriin..10{ handles.push(my_bg_task(i));// 没有把 Future 变成任务 } std::thread::sleep(Duration::from_millis(120)); println!("Finished time-consuming task."); futures::future::join_all(handles).await;// 但是 join_all 会等待所有 Future 并发执行完 println!("总耗时：{} ms", now.elapsed().as_millis()); } asyncfnmy_bg_task(i:u64) { letmillis =100; println!("Task {} sleeping for {} ms.", i, millis); sleep(Duration::from_millis(millis)).await; println!("Task {} stopping.", i); }

运行结果：异步阻塞
 

Finishedtime-consuming task. Tasksleepingfor100ms. Task1sleepingfor100ms. Task2sleepingfor100ms. Task3sleepingfor100ms. Task4sleepingfor100ms. Task5sleepingfor100ms. Task6sleepingfor100ms. Task7sleepingfor100ms. Task8sleepingfor100ms. Task9sleepingfor100ms. Taskstopping. Task1stopping. Task2stopping. Task3stopping. Task4stopping. Task5stopping. Task6stopping. Task7stopping. Task8stopping. Task9stopping.

总耗时：221ms

P.S. 关于代码中 和 的区别，参考这篇文章 Async: What is blocking? (by Alice Ryhl) 。