Rust中Async move闭包与fold的对比及问题咨询

问题 #

Rust中的async闭包目前仍处于unstable状态，但我对相关解答不甚理解。我认为以下代码并非async闭包：

let mut sum: i32 = 0;
stream::iter(1..25)
    .map(compute)
    .buffered(12)
    .for_each(|result| async move { sum+=result; })
    .await;
println!("->{}", sum);

起初我感到困惑：sum在for_each中被使用却未被转移（否则println!会触发编译错误），但编译器提示sum的赋值未被读取，实际上sum被复制了。完整示例代码如下：

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

async fn compute(i: i32) -> i32 {
    let mut rng = thread_rng();
    let sleep_ms: u64 = rng.gen_range(0..1000);
    tokio::time::sleep(Duration::from_millis(sleep_ms)).await;
    println!("#{} done", i);
    i * i
}

async fn sum_with_fold() {
    let sum = stream::iter(1..25)
        .map(compute)
        .buffered(12)
        .fold(0, |sum,x| async move {sum+x} )
        .await;
    println!("->{}", sum);
}

async fn sum_with_closure() {
    let mut sum: i32 = 0;
    stream::iter(1..25)
        .map(compute)
        .buffered(12)
        .for_each(|result| async move { sum+=result; })
        .await;
    println!("->{}", sum);
}

#[tokio::main]
async fn main() {
    sum_with_fold().await;
    sum_with_closure().await;
}

// Cargo.toml:
// [dependencies]
// futures = "0.3"
// rand = "0.8"
// tokio = { version = "1", features = ["full"] }

其中fold能正确计算总和，而sum_with_closure中操作的是sum的副本，无法获取正确结果。我的理解是否正确？能否修复该问题？是否可以用类似闭包实现fold的功能？这是否涉及到不稳定的async闭包特性？

解答 #

1. 你的理解完全正确 #

sum_with_closure里的问题在于：

• async move块会捕获变量的所有权，但i32是Copy类型，所以实际上是复制了一份sum到async块内部，外部的sum仍然是初始值0。
• 你修改的只是async块里的副本，外部的sum根本没被改动，所以编译器会提示“赋值未被读取”——因为副本的修改没有任何后续使用场景。

2. 修复方法：使用共享可变状态 #

由于buffered(12)会同时并发处理多个元素，多个async块可能同时修改sum，所以需要线程安全的共享可变容器。推荐使用Arc<Mutex<i32>>：

use std::sync::{Arc, Mutex};

async fn sum_with_closure_fixed() {
    let sum = Arc::new(Mutex::new(0));
    stream::iter(1..25)
        .map(compute)
        .buffered(12)
        .for_each(|result| {
            let sum = Arc::clone(&sum);
            async move {
                let mut inner = sum.lock().unwrap();
                *inner += result;
            }
        })
        .await;
    println!("->{}", *sum.lock().unwrap());
}

如果是单线程运行时（比如tokio指定flavor="current_thread"），也可以用RefCell替代Mutex，避免线程同步开销，但Mutex是更通用的线程安全方案。

3. 用闭包实现类似fold的功能 #

fold本身就是通过闭包实现的累加逻辑，它会按顺序处理流中的元素，将上一次的结果传递给下一次闭包调用，天然避免了共享可变状态的问题。如果非要用for_each模拟，本质上就是用共享状态来手动实现fold的累加，就像上面的修复代码那样——但这不如直接用fold简洁安全。

4. 与不稳定async闭包的关系 #

你当前的代码不涉及不稳定的async闭包特性。当前写法是“普通闭包返回async块”，这是稳定的语法。而真正的async闭包是async |result| { ... }这种直接标记为async的闭包，目前仍处于Rust的unstable阶段，需要启用async_closure特性才能使用。

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