异步 Rust 中的多线程 - 为什么我的代码无法并行化？

Question

我试图通过运行以下函数来有意耗尽 API 限制（900 次调用）：

#[get("/exhaust")]
pub async fn exhaust(_pool: web::Data<PgPool>, config: web::Data<Arc<Settings>>) -> impl Responder {
    let mut handles = vec![];

    for i in 1..900 {
        let inner_config = config.clone();
        let handle = thread::spawn(move || async move {
            println!("running thread {}", i);
            get_single_tweet(inner_config.as_ref().deref(), "1401287393228038149")
                .await
                .unwrap();
        });
        handles.push(handle);
    }

    for h in handles {
        h.join().unwrap().await;
    }

    HttpResponse::Ok()

我的机器有 16 个内核，所以我希望上面的运行速度比单线程函数快 16 倍，但事实并非如此。事实上，它的运行速度与单线程版本一样慢。

这是为什么？我错过了什么？

注意：这move || async move部分对我来说看起来有点奇怪，但我是按照编译器的建议到达那里的。由于async closures being unstable. 这可能是问题吗？

Answer 1

问题是您混合多线程和异步的方式导致所有工作都是顺序的：您的所有线程所做的就是调用get_single_tweet这显然是一个async函数。

现在，在像 Javascript 这样的语言中，get_single_tweet将创建一个任务，该任务将返回一个承诺，象征着任务的实现并尽快运行。

这不是 Rust 的工作方式（或者许多其他语言，顺便说一句，Python 的行为比 Javascript 更像 Rust）。在 Rust 中，get_single_tweet只是创建一个future，它实际上并没有做任何事情，必须对 future 进行轮询才能发生事情：https://play.rust-lang.org/ ?version=stable&mode=debug&edition=2018&gist=b26b47e62e46b66b60844aabc2ea7be1

这次投票什么时候进行？当等待的动态链到达事件循环的顶部时。

在这里，未来是在线程中创建的，然后从线程返回，然后await在从 中获取时进行 -ed join，因此您的获取不会在线程中运行，而是在此处运行：

    for h in handles {
        h.join().unwrap().await;
    }

这是完全连续的。

现在要解决这个问题，您基本上有两个选择，但重要的是要保持在您的范围内：

• 如果你想使用线程，那么线程中的调用应该是阻塞的
• 否则，不要使用thread::spawn您可能想要使用的task::spawn（或您选择的运行时中的任何等效项），确保您使用多线程运行时可能是一个好主意，但在这里我假设get_single_tweet正在执行 IO 工作（例如获取来自 twitter 的东西），因此单线程运行时可能已经产生了大部分好处

task::spawn将（如其名称所示）创建一个任务并返回它的句柄，这更接近 Javascriptasync函数的行为。

Answer 2

此代码确实会async同步运行您的块。一个async块创建了一个实现 的类型Future，但要知道的一件事是Futures 不会自己开始运行，它们必须被await-ed 或提供给执行程序才能运行。

thread::spawn使用返回 a 的闭包调用Future不会执行它们；线程只是创建async块并返回。所以这些async块实际上并没有被执行，直到你await在循环中它们才会被执行handles，这将按顺序处理期货。

解决此问题的一种方法是join_all从futures板条箱中同时运行它们。

let mut futs = vec![];

for i in 1..900 {
    let inner_config = config.clone();
    futs.push(async move {
        println!("running thread {}", i);
        get_single_tweet(inner_config.as_ref().deref(), "1401287393228038149")
            .await
            .unwrap();
    });
}

futures::future::join_all(futs).await;