如何使用reqwest执行并行异步HTTP GET请求？

Question

异步示例很有用,但作为Rust和Tokio的新手,我正在努力研究如何一次执行N个请求,使用向量中的URL,并为每个URL创建响应HTML的迭代器作为字符串.

怎么可以这样做？

Answer 1

截至reqwest 0.9:

use futures::{stream, Future, Stream}; // 0.1.26
use reqwest::r#async::Client; // 0.9.14
use tokio; // 0.1.18

type Result<T> = std::result::Result<T, Box<std::error::Error>>;

const PARALLEL_REQUESTS: usize = 2;

fn main() -> Result<()> {
    let client = Client::new();

    let urls = vec!["https://api.ipify.org", "https://api.ipify.org"];

    let bodies = stream::iter_ok(urls)
        .map(move |url| {
            client
                .get(url)
                .send()
                .and_then(|res| res.into_body().concat2().from_err())
        })
        .buffer_unordered(PARALLEL_REQUESTS);

    let work = bodies
        .for_each(|b| {
            println!("Got {} bytes", b.len());
            Ok(())
        })
        .map_err(|e| panic!("Error while processing: {}", e));

    tokio::run(work);
    Ok(())
}

stream::iter_ok(urls)

stream::iter_ok

获取字符串集合并将其转换为Stream.

.and_then(|res| res.into_body().concat2().from_err())

Stream::concat2, Stream::from_err

获取每个响应的正文流并将其全部收集到一个大块中.

.buffer_unordered(N);

Stream::buffer_unordered

将期货流转换为未来价值流,并行执行期货.

let work = bodies.for_each(|b| {
    println!("Got {} bytes", b.len());
    Ok(())
});

Stream::for_each

将流转换回未来.

也可以看看:

• 加入有限并发的期货
• 如何合并流的迭代器？
• 如何在稳定的Rust中同步返回异步Future中计算的值？