Rust如何在crossbeam scope线程的spawn中执行async异步函数

错误原因 #

• 第一个编译错误的核心是：crossbeam_utils::thread::scope 创建的是普通同步操作系统线程，s.spawn 接收的是同步闭包，而 .await 只能在 async 函数/async 块内部使用，同步闭包没有async上下文自然不支持await语法。
• 后续给闭包加async标识后代码不执行的原因：async闭包执行后只会返回一个实现了Future trait的对象，Future本身是惰性的，没有异步运行时驱动执行的话，内部代码永远不会运行。

解决方案 #

你需要在crossbeam创建的同步线程内部，启动一个轻量异步运行时来驱动test函数对应的Future执行，以常用的Tokio运行时为例：

第一步：添加依赖 #

在Cargo.toml中加入Tokio依赖，开启轻量单线程运行时特性：

[dependencies]
tokio = { version = "1.0", features = ["rt"] }

第二步：修改代码 #

在s.spawn的同步闭包内创建单线程运行时，用block_on方法执行async逻辑：

use crossbeam_utils::thread;
use core_lib::request;
use tokio::runtime;

pub async fn test(){
    // Will perform stuff here
    println!("will perform stuff here");
}

pub async fn initiate(){
    if let Ok(result) = request::Request::get_request().await{
        // 注意这里改成into_iter拿走所有权，否则后面move r的时候会有生命周期问题
        let message = result.message().into_iter();
        thread::scope(|s| {
            for r in message{
                log::debug!("Detected request id: {}",r.id());
                println!("{:?}",r.config());

                s.spawn(move |_| {
                    log::info!("Processing request id: {}", r.id());
                    // 创建当前线程的异步运行时
                    let rt = runtime::Builder::new_current_thread()
                        .build()
                        .unwrap();
                    // 驱动Future执行直到完成
                    rt.block_on(test());
                });
            }
        }).unwrap_or_default();
    }
}

可选优化 #

如果你的业务逻辑以IO密集型为主，不需要单独开操作系统线程隔离任务，完全可以直接用异步运行时自带的spawn方法，不需要结合crossbeam线程，性能更好：

pub async fn initiate(){
    if let Ok(result) = request::Request::get_request().await{
        let message = result.message().into_iter();
        let mut handles = Vec::new();
        for r in message{
            log::debug!("Detected request id: {}",r.id());
            println!("{:?}",r.config());
            // 直接用tokio的spawn提交异步任务
            let handle = tokio::spawn(async move{
                log::info!("Processing request id: {}", r.id());
                test().await;
            });
            handles.push(handle);
        }
        // 等待所有异步任务执行完成，和crossbeam scope逻辑对齐
        for handle in handles {
            let _ = handle.await;
        }
    }
}

如果你使用的是async-std运行时，替换成async_std::task::block_on即可，不需要手动创建运行时。

内容的提问来源于stack exchange，提问作者Kheshav Sewnundun