什么时候使用FuturesOrdered?

对于同一个函数,我有两种方法,我正在尝试看哪一种更惯用或更具表演性。

query_many_points() 函数是接收两个相同大小的数组,并通过 query_point() 函数,返回的向量 Response s-包含从API反序列化的数据的自定义结构。API响应的顺序必须与输入的顺序完全匹配。我相信这两种方法都能做到这一点。

我最初是通过在Vec中收集期货来写这篇文章的,然后迭代到 await 并按顺序展开结果。

方法1:把手的V形

fn query_many_points(xs: &[f64], ys: &[f64]) -> anyhow::Result<Vec<Response>> {
    let agent = ureq::agent();
    let semaphore = Arc::new(Semaphore::new(TASKS_LIMIT));

    // Concurrently query all given coordinates, limited to TASKS_LIMIT concurrent tasks
    // Returned Vec<Response> preserves order of input coordinates
    let runtime = Runtime::new()?;
    runtime.block_on(async {
        // Create a vector of task handles, preserving order with input coordinates
        let handles = xs
            .iter()
            .zip(ys.iter())
            .map(|(&x, &y)| {
                // Obtain permit from semaphore when available
                let permit = semaphore.clone().acquire_owned();
                let agent = agent.clone();
                // Spawn new task and query point, returning handle containing Result<Response>
                tokio::spawn(async move {
                    let result = query_point(x, y, &agent);
                    drop(permit);
                    result
                })
            })
            .collect::<Vec<_>>(); // this line is where the approaches start to differ

        // Await all tasks to complete in order and collect into Vec<Response>
        let mut responses = Vec::with_capacity(handles.len());
        for handle in handles {
            responses.push(handle.await??);
        }

        Ok(responses)
    })
}

然后我发现 futures::stream::FuturesOrdered 然后转了一圈,想出了第二个解决方案。

方法2:未来有序

fn query_many_points(xs: &[f64], ys: &[f64]) -> anyhow::Result<Vec<Response>> {
    let agent = ureq::agent();
    let semaphore = Arc::new(Semaphore::new(TASKS_LIMIT));

    // Concurrently query all given points, limited to TASKS_LIMIT concurrent tasks
    // Returned Vec<Response> preserves order of input points
    let runtime = Runtime::new()?;
    runtime.block_on(async {
        // Create a FuturesOrdered of task handles
        let handles = xs
            .iter()
            .zip(ys.iter())
            .map(|(&x, &y)| {
                // Obtain permit from semaphore when available
                let permit = semaphore.clone().acquire_owned();
                let agent = agent.clone();
                // Spawn new task and query point, returning handle containing Result<Response>
                tokio::spawn(async move {
                    let result = query_point(x, y, &agent);
                    drop(permit);
                    result
                })
            })
            .collect::<FuturesOrdered<_>>(); // this line is where the approaches start to differ

        // Await completion of all tasks in order and collect into Vec<Response>
        handles.try_collect::<Vec<_>>().await?.into_iter().collect()
    })
}

哪种方法最好?还有其他可以改进的地方吗?两者保持秩序完全相同,对吧?