前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >【原创】Rust tokio 如何以异步非阻塞方式运行大量任务

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

作者头像
MikeLoveRust
发布2021-09-08 10:11:21
4.5K0
发布2021-09-08 10:11:21
举报
文章被收录于专栏:Rust语言学习交流

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

而下面利用 #[tokio::main] 和 await 编写了等价的版本(为了直观对比任务完成的实际顺序和总耗时,我对 sleep 的时间做了一些简化):

代码语言:javascript
复制
use std::time::Instant;
use tokio::time::{sleep, Duration};

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let now = Instant::now();

    let mut handles = Vec::with_capacity(10);
    for i in 0..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.
    for handle in handles {
        handle.await?;
    }

    println!("总耗时:{} ms", now.elapsed().as_millis());
    Ok(())
}

async fn my_bg_task(i: u64) {
    let millis = 100;
    println!("Task {} sleeping for {} ms.", i, millis);
    sleep(Duration::from_millis(millis)).await;
    println!("Task {} stopping.", i);
}

输出结果:

代码语言:javascript
复制
Task 0 sleeping for 100 ms.
Task 1 sleeping for 100 ms.
Task 2 sleeping for 100 ms.
Task 3 sleeping for 100 ms.
Task 4 sleeping for 100 ms.
Task 5 sleeping for 100 ms.
Task 6 sleeping for 100 ms.
Task 7 sleeping for 100 ms.
Task 8 sleeping for 100 ms.
Task 9 sleeping for 100 ms.
Task 9 stopping.
Task 0 stopping.
Task 1 stopping.
Task 2 stopping.
Task 3 stopping.
Task 4 stopping.
Task 5 stopping.
Task 6 stopping.
Task 7 stopping.
Task 8 stopping.
Finished time-consuming task.
总耗时:120 ms

如果把主线程的的 sleep 时间改成 100 ms:std::thread::sleep(Duration::from_millis(100)); 则产生下面的结果:

代码语言:javascript
复制
Task 0 sleeping for 100 ms.
Task 1 sleeping for 100 ms.
Task 2 sleeping for 100 ms.
Task 3 sleeping for 100 ms.
Task 4 sleeping for 100 ms.
Task 5 sleeping for 100 ms.
Task 6 sleeping for 100 ms.
Task 7 sleeping for 100 ms.
Task 8 sleeping for 100 ms.
Task 9 sleeping for 100 ms.
Finished time-consuming task.
Task 3 stopping.
Task 0 stopping.
Task 1 stopping.
Task 2 stopping.
Task 9 stopping.
Task 4 stopping.
Task 5 stopping.
Task 6 stopping.
Task 7 stopping.
Task 8 stopping.
总耗时:103 ms

可以看到,my_bg_task 实际是异步非阻塞执行的 👍 :

  • 异步:因为每个任务不必等待其结果就可以开始下一个任务,即;
代码语言:javascript
复制
// 异步
Task 0 sleeping for 100 ms.
Task 1 sleeping for 100 ms.
...

// 同步
Task 0 sleeping for 100 ms.
Task 0 stopping.
Task 1 sleeping for 100 ms.
Task 1 stopping.
...
  • 非阻塞:每个任务之间可以快速切换,不必等待其他任务完成才切换,这个例子表现在:
    • 任务 0-9 以乱序方式 stop
    • Finished time-consuming task.Task x stopping. 的打印顺序只与任务各自的运行 (sleep) 时间有关,与源代码的声明执行顺序无关。只有任务之间快速切换才能做到这一点。回顾官网的例子:10 个任务的 sleep 时间线性递减 (let millis = 1000 - 50 * i;),从 6 个任务开始小于主线程 sleep 任务的时间(750 ms),而等待 10 个任务执行的语句 for handle in handles { ... } 显然位于 std::thread::sleep 之后,所以任务之间非阻塞执行的话,打印结果为 sleep 时间越短的任务先完成,时间越长的任务后完成,总耗时为任务中的最长耗时:
代码语言:javascript
复制
Task 0 sleeping for 1000 ms.
Task 1 sleeping for 950 ms.
Task 2 sleeping for 900 ms.
Task 3 sleeping for 850 ms.
Task 4 sleeping for 800 ms.
Task 5 sleeping for 750 ms.
Task 6 sleeping for 700 ms.
Task 7 sleeping for 650 ms.
Task 8 sleeping for 600 ms.
Task 9 sleeping for 550 ms.
Task 9 stopping.
Task 8 stopping.
Task 7 stopping.
Task 6 stopping.
Finished time-consuming task.
Task 5 stopping.
Task 4 stopping.
Task 3 stopping.
Task 2 stopping.
Task 1 stopping.
Task 0 stopping.
总耗时:1001 ms // 非常完美

一般情况下,对于 async block/fn 你至少有以下一些做法:

  1. 对 async block/fn 调用 .await 来等待结果;
  2. 对可列举的少数 Future 调用 join! 或者 select! 来同时等待多个结果 或者 等待多个分支的第一个结果;
  3. 对大量 Future 调用 join 或者 select 一类支持传入 Vec / iter 参数类型的函数,比如这个例子中的 for handle in handles { ... } 部分就可以改写成 futures::future::join_all(handles).await;
  4. 把 async block/fn 变成任务,然后调用 Runtime::block_on (等价地,对任务 await)来执行许多任务。

容易犯的错误是,希望异步非阻塞时,对所有 async block/fn 进行了 await,而没有进行任务化处理(即 把 Future 通过 spwan 函数转化成任务):

代码语言:javascript
复制
use std::time::Instant;
use tokio::time::{sleep, Duration};

#[tokio::main]
async fn main() {
    let now = Instant::now();

    let mut handles = Vec::with_capacity(10);
    for i in 0..10 {
        handles.push(my_bg_task(i)); // 没有把 Future 变成任务
    }

    std::thread::sleep(Duration::from_millis(120));
    println!("Finished time-consuming task.");

    for handle in handles {
        handle.await; // 而且每个 handle 必须执行完才能执行下一个 handle
    }
    println!("总耗时:{} ms", now.elapsed().as_millis());
}

async fn my_bg_task(i: u64) {
    let millis = 100;
    println!("Task {} sleeping for {} ms.", i, millis);
    sleep(Duration::from_millis(millis)).await;
    println!("Task {} stopping.", i);
}

运行结果:同步阻塞

代码语言:javascript
复制
Finished time-consuming task.
Task 0 sleeping for 100 ms.
Task 0 stopping.
Task 1 sleeping for 100 ms.
Task 1 stopping.
Task 2 sleeping for 100 ms.
Task 2 stopping.
Task 3 sleeping for 100 ms.
Task 3 stopping.
Task 4 sleeping for 100 ms.
Task 4 stopping.
Task 5 sleeping for 100 ms.
Task 5 stopping.
Task 6 sleeping for 100 ms.
Task 6 stopping.
Task 7 sleeping for 100 ms.
Task 7 stopping.
Task 8 sleeping for 100 ms.
Task 8 stopping.
Task 9 sleeping for 100 ms.
Task 9 stopping.
总耗时:1130 ms

或者像这样:

代码语言:javascript
复制
use std::time::Instant;
use tokio::time::{sleep, Duration};

#[tokio::main]
async fn main() {
    let now = Instant::now();

    let mut handles = Vec::with_capacity(10);
    for i in 0..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());
}

async fn my_bg_task(i: u64) {
    let millis = 100;
    println!("Task {} sleeping for {} ms.", i, millis);
    sleep(Duration::from_millis(millis)).await;
    println!("Task {} stopping.", i);
}

运行结果:异步阻塞

代码语言:javascript
复制
Finished time-consuming task.
Task 0 sleeping for 100 ms.
Task 1 sleeping for 100 ms.
Task 2 sleeping for 100 ms.
Task 3 sleeping for 100 ms.
Task 4 sleeping for 100 ms.
Task 5 sleeping for 100 ms.
Task 6 sleeping for 100 ms.
Task 7 sleeping for 100 ms.
Task 8 sleeping for 100 ms.
Task 9 sleeping for 100 ms.
Task 0 stopping.
Task 1 stopping.
Task 2 stopping.
Task 3 stopping.
Task 4 stopping.
Task 5 stopping.
Task 6 stopping.
Task 7 stopping.
Task 8 stopping.
Task 9 stopping.
总耗时:221 ms

P.S. 关于代码中 std::thread::sleeptokio::time::sleep 的区别,参考这篇文章 Async: What is blocking? (by Alice Ryhl) 。

本文参与 腾讯云自媒体同步曝光计划,分享自微信公众号。
原始发表:2021-09-01,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 Rust语言学习交流 微信公众号,前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档