rust - 使用新的异步/等待语法发送多组 HTTP 请求并控制工作人员数量的最快方法

Question

我最近读到的帖子都说异步是执行大量 I/O 绑定(bind)工作(例如发送 HTTP 请求等)的更好方法。我最近尝试使用异步，但在理解如何并行发送多组请求方面遇到困难，例如:

let client = reqwest::Client::new();
let mut requests = 0;

let get = client.get("https://somesite.com").send().await?;
let response = get.text().await?;

if response.contains("some stuff") {
    let get = client.get("https://somesite.com/something").send().await?;
    let response = get.text().await?;

    if response.contains("some new stuff") {
        requests += 1;
        println!("Got response {}", requests)

这符合我的要求，但我如何才能并行运行它并控制“工作线程”的数量或异步线程池的等价物？

我理解它类似于 this question ，但我的严格来说是在谈论每晚的 Rust async/await 语法和一个更具体的用例，其中需要完成一组请求/任务。我还发现在这些情况下使用组合器有点令人困惑，希望更新的样式有助于使其更具可读性。

Answer 1

不确定这是否是最快的方法，因为我只是在试验自己，但这是我的解决方案:

let client = reqwest::Client::new();

let links = vec![ // A vec of strings representing links
    "example.net/a".to_owned(), 
    "example.net/b".to_owned(),
    "example.net/c".to_owned(),
    "example.net/d".to_owned(),
    ];

let ref_client = &client; // Need this to prevent client from being moved into the first map
futures::stream::iter(links)
    .map(async move |link: String| {
        let res = ref_client.get(&link).send().await;

        // res.map(|res| res.text().await.unwrap().to_vec())
        match res { // This is where I would usually use `map`, but not sure how to await for a future inside a result 
            Ok(res) => Ok(res.text().await.unwrap()),
            Err(err) => Err(err), 
        }
    })
    .buffer_unordered(10) // Number of connection at the same time
    .filter_map(|c| future::ready(c.ok())) // Throw errors out, do your own error handling here
    .filter_map(|item| {
        if item.contains("abc") {
            future::ready(Some(item))
        } else {
            future::ready(None)
        }
    })
    .map(async move |sec_link| {
        let res = ref_client.get(&sec_link).send().await;
        match res {
            Ok(res) => Ok(res.text().await.unwrap()),
            Err(err) => Err(err),
        }
    })
    .buffer_unordered(10) // Number of connections for the secondary requests (so max 20 connections concurrently)
    .filter_map(|c| future::ready(c.ok()))
    .for_each(|item| {
        println!("File received: {}", item);
        future::ready(())
    })
    .await;

这需要 #![feature(async_closure)] 特性。