Getting multiple URLs concurrently with Hyper

问题:

I am trying to adapt the Hyper basic client example to get multiple URLs concurrently.

This is the code I currently have:

extern crate futures;
extern crate hyper;
extern crate tokio_core;

use std::io::{self, Write};
use std::iter;
use futures::{Future, Stream};
use hyper::Client;
use tokio_core::reactor::Core;

fn get_url() {
    let mut core = Core::new().unwrap();
    let client = Client::new(&core.handle());
    let uris: Vec<_> = iter::repeat("http://httpbin.org/ip".parse().unwrap()).take(50).collect();
    for uri in uris {
        let work = client.get(uri).and_then(|res| {
            println!("Response: {}", res.status());

            res.body().for_each(|chunk| {
                io::stdout()
                    .write_all(&chunk)
                    .map_err(From::from)
            })
        });
        core.run(work).unwrap();
    }
}

fn main() {
    get_url();
}

It doesn't seem to be acting concurrently (it takes a long time to complete), am I giving the work to the core in the wrong way?

回答1:

am I giving the work to the core in the wrong way?

Yes, you are giving one request to Tokio and requiring that it complete before starting the next request. You've taken asynchronous code and forced it to be sequential.

You need to give the reactor a single future that will perform different kinds of concurrent work:

use futures::{stream, Future, Stream}; // 0.1.25
use hyper::Client; // 0.12.23
use std::{
    io::{self, Write},
    iter,
};
use tokio; // 0.1.15

const N_PARALLEL: usize = 1;

fn main() {
    let client = Client::new();

    let uri = "http://httpbin.org/ip".parse().unwrap();
    let uris = iter::repeat(uri).take(50);

    let work = stream::iter_ok(uris)
        .map(move |uri| client.get(uri))
        .buffer_unordered(N_PARALLEL)
        .and_then(|res| {
            println!("Response: {}", res.status());
            res.into_body()
                .concat2()
                .map_err(|e| panic!("Error collecting body: {}", e))
        })
        .for_each(|body| {
            io::stdout()
                .write_all(&body)
                .map_err(|e| panic!("Error writing: {}", e))
        })
        .map_err(|e| panic!("Error making request: {}", e));

    tokio::run(work);
}

With N_PARALLEL set to 1:

real    0m2.279s
user    0m0.193s
sys     0m0.065s

And set to 10:

real    0m0.529s
user    0m0.186s
sys     0m0.075s