如何从同时工作的for循环中收集多个结果？

Question

我想运行一个包含许多不同变量的函数

假设我的功能是：

async fn do_the_hard_job(my_input: u16) {
    ...
    // do the hard job that takes some time
    if my_condition {
        println!("{}", input);
    }
}

我想为许多输入运行这个函数；假设我的输入是

inputs = stream::iter(1..=u16::MAX);

由于该函数会花费一些周期，因此我希望尽可能同时运行所有输入。所以我可以这样运行

 inputs.for_each_concurrent(0, |input| do_the_hard_job(input))
       .await;

到目前为止一切顺利；我使用该函数运行所有输入并在stdout 上获得我的输出。但是如果我想将输出写入某个输出文件呢？

我无法在do_the_hard_job 函数中打开文件并将其附加到其中。那会搞砸的。我无法将文件添加为参数，因为该方法将同时运行，哪个将借用可变文件。

我尝试返回值而不是在方法中打印，然后收集返回的值；像这样：

let mut return_values:Vec<u16> = Vec::new();
inputs
    .for_each_concurrent(0, |input| async move {
        if let done = do_the_hard_job(port).await{
            if done > 0 {
                return_values.push(port);
            }
        }}).await;

很遗憾，这没有用。我可以尝试什么来实现我的目标？

编辑：我为这个问题准备了一个复制器：https://github.com/kursatkobya/bite-sized-qualms/tree/main/concurrent-write

Answer 1

你可以结合then和collect得到结果：

use futures::{stream, StreamExt};
use std::{
    time::Duration,
};

#[tokio::main]
async fn main() {
    let inputs = stream::iter(0..=10);

    // Finaly the one below does not work
    let output_values: Vec<u16> = inputs
        .then(|input: u16| async move {
            let result = f(input).await;
            result
        }).collect::<Vec<u16>>().await;
    println!("{:?}", output_values);

}

async fn f(input: u16) -> u16 {
    tokio::time::sleep(Duration::from_millis(200)).await;
    input
}

Playground