Rust中buffered future内给Vec追加元素报移动错误如何解决

代码编译时抛出如下所有权错误：

cannot move out of `my_number_vector`, a captured variable in an `FnMut` closure
move out of `my_number_vector`

依赖配置（Cargo.toml） #

[dependencies]
futures = "0.3.21"
tokio = "1.19.2"

问题代码 #

use futures::{stream, StreamExt};


#[tokio::main]
async fn main() {

let mut my_number_vector = Vec::new();

let the_stream = stream::iter(0..=10);


let response_of_stream = stream_of_urls // 存在笔误：变量名应为 the_stream
            .map(|number| async move {
              number
            })
            .buffer_unordered(2);

 response_of_stream
            .for_each(|number| async move {
               my_number_vector.push(number);
            })
            .await;

}

核心问题有三个：

• 代码存在低级笔误：定义的流变量名为the_stream，后续调用时错写为不存在的stream_of_urls，会直接触发变量不存在的编译错误。
• 所有权逻辑错误：for_each要求传入的闭包为FnMut类型，支持被反复调用；但闭包内写了async move，第一次执行时就会把捕获的my_number_vector所有权移动到当前异步块中，后续闭包再被调用时，已经没有该变量的所有权可用，因此触发移动错误。
• 之前尝试Rc<RefCell>方案失败的原因是：Rc是单线程引用计数类型，未实现Send trait，无法在并发任务、跨.await点传递；RefCell的运行时借用检查也不支持线程安全，在buffer_unordered的并发执行场景下会出现数据竞争，无法通过编译。

方案1：直接用collect收集结果（最推荐） #

如果只是要把流的所有输出存入Vec，不需要在遍历中加其他逻辑，直接调用StreamExt提供的collect方法即可，不需要手动维护Vec、处理所有权问题，代码最简洁：

use futures::{stream, StreamExt};

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

    // 直接将流执行结果收集为Vec，自动处理元素顺序和所有权
    let my_number_vector: Vec<i32> = the_stream
            .map(|number| async move { number })
            .buffer_unordered(2)
            .collect()
            .await;

    println!("收集结果：{:?}", my_number_vector);
}

方案2：用Arc包裹集合（适合遍历中需要额外逻辑的场景） #

如果必须在for_each遍历过程中执行push操作，同时适配buffer_unordered的并发场景，需要用线程安全的引用计数Arc加异步互斥锁tokio::sync::Mutex包裹Vec，既可以让多个异步任务共享Vec的所有权，也能保证并发修改的安全性：

use futures::{stream, StreamExt};
use std::sync::Arc;
use tokio::sync::Mutex;

#[tokio::main]
async fn main() {
    // 用Arc<Mutex>包裹Vec，支持跨任务共享所有权、安全并发修改
    let my_number_vector = Arc::new(Mutex::new(Vec::new()));
    let the_stream = stream::iter(0..=10);

    let response_of_stream = the_stream
            .map(|number| async move { number })
            .buffer_unordered(2);

    response_of_stream
            .for_each(|number| {
                // 每次调用仅克隆Arc的引用计数，不移动底层Vec
                let vec_handle = Arc::clone(&my_number_vector);
                async move {
                    // 加锁后执行push，避免并发写冲突
                    vec_handle.lock().await.push(number);
                }
            })
            .await;

    // 所有任务执行完成后，取出Arc内部的Vec
    let final_vec = Arc::into_inner(my_number_vector).unwrap().into_inner();
    println!("收集结果：{:?}", final_vec);
}

内容的提问来源于stack exchange，提问作者Corfucinas