以有限的并发性加入期货

Question

我有一个很大的超HTTP请求期货向量，并希望将它们解析为结果向量。由于存在最大打开文件的限制，所以我希望将并发限制为N个期货。

我试验过Stream::buffer_unordered，但它似乎是一个接一个地执行期货。

Answer 1

我们在一个项目中使用了这样的代码，以避免打开太多的TCP套接字。这些期货里面有超期货，所以情况似乎完全一样。

// Convert the iterator into a `Stream`. We will process
// `PARALLELISM` futures at the same time, but with no specified
// order.
let all_done =
    futures::stream::iter(iterator_of_futures.map(Ok))
    .buffer_unordered(PARALLELISM);

// Everything after here is just using the stream in
// some manner, not directly related

let mut successes = Vec::with_capacity(LIMIT);
let mut failures = Vec::with_capacity(LIMIT);

// Pull values off the stream, dividing them into success and
// failure buckets.
let mut all_done = all_done.into_future();
loop {
    match core.run(all_done) {
        Ok((None, _)) => break,
        Ok((Some(v), next_all_done)) => {
            successes.push(v);
            all_done = next_all_done.into_future();
        }
        Err((v, next_all_done)) => {
            failures.push(v);
            all_done = next_all_done.into_future();
        }
    }
}

这在一段示例代码中使用，因此事件循环(core)是显式驱动的。查看该程序使用的文件句柄的数量显示它是有上限的。此外，在添加此瓶颈之前，我们很快就用完了允许的文件句柄，而之后却没有。