当前位置：首页 > article >正文

Rust使用tokio（一）

article 2025/1/26 17:51:52

tokio中的task，是tokio核心调度的基本单位，位于操作系统线程之上，所以又被称为绿色线程，类似于Go语言中的goroutines、Kotline里面的coroutines或者Erlang里的processes。

task是轻量的，非阻塞的，当遇到系统IO等阻塞事件时，tokio会切换到其它的task执行。

spawn

最常用的task相关的API就是spawn，它的使用方法是：

use tokio::task;

task::spawn(async {
    // perform some work here...
});

即只有一个参数，参数类型是一个Future，返回值是一个JoinHandle。

当执行完毕之后，JoinHandle会返回原函数的返回值，或者当遇到panic的时候，返回一个JoinError。

JoinHandle

JoinHandle支持取消，方法是调用abort()函数。

当调用JoinHandle的abort()函数之后，这个Future的处理，将分为两种情况。

当前Future处于执行状态时，则记录这个中止状态，等执行到await的时候，退出执行。
当前Future处于非执行状态，即idle时，则马上退出执行。

阻塞执行

在tokio中的task是非阻塞操作，另外执行的函数也不要有阻塞的调用，因为阻塞的调用，会造成其它的task无法并行执行。

但是，如果确实需要阻塞执行，可以使用task::spawn_blocking或者task::block_in_place函数。

task::spawn_blocking将把参数指定的过程，放在一个单独的线程中执行，然后也返回一个JoinHandle。可以通过这个JoinHandle获得程序最后执行的结果。
task::block_in_place将把参数指定的过程，放在当前的线程中执行，把这个线程上的其它task都移到其它线程去。

tokio::main

使用tokio框架的时候，可以把main函数包装成一个tokio运行时。方法是把main实现成async，同时加上宏tokio::main。

#[tokio::main]
async fn main() {
    let args = std::env::args().collect::<Vec<String>>();
    for arg in args[1..].iter() {
        println!("{}", arg);
    }
}

tokio::test

另外如果要编写测试用例的话，也需要在函数实现为async，同时加上tokio::test宏。

如：

mod tests {  
    use super::*; 
 
    #[tokio::test]  
    async fn test_mongo_connect() {  
        let uri = format!("mongodb://{}:{}", "localhost", 12306);
        let client = mongodb::Client::with_uri_str(&uri).await;
        assert_eq!(ret.is_ok(), true);  
    }  
}

async-trait

tokio是异步的，所以使用这个库的时候，很多函数都得是async的。

而当我们定义trait的时候，默认却是不能包含async函数的。解决方法是使用async-trait。

async-trait需要我们手动加入依赖，之后在需要async方法的trait前面加上async_trait宏。

如：

use async_trait::async_trait;

#[async_trait]
pub trait DBConnect {
    async fn connect(&mut self, host: &str, port: u16, db: &str) -> Result<bool, String>;

之后，在相应的实现中，也在前面加上这个宏：

use async_trait::async_trait;
use mongodb;

#[async_trait]
impl DBConnect for MongoDB {
    async fn connect(&mut self, host: &str, port: u16, db: &str) -> Result<mongodb::Database, String> {
         let uri = format!("mongodb://{}:{}", host, port);
        let client = mongodb::Client::with_uri_str(&uri).await;
        match client {
            Ok(client) => {
                Ok(client.database(db))
            }
            Err(e) => {
                Err(format!("{}", e))
            }
        }
    }
}