Rust异步更新Polars DataFrame时遇`FnMut`闭包变量逃逸错误

我尝试用Rust编写代码，在异步获取msg后通过拼接操作更新Polars库的主DataFrame df。此前用字符串执行类似更新操作正常，但当前代码报错，已查阅Stack Overflow相关帖子仍未明确问题原因。我仅希望可变借用DataFrame并完成更新，相关代码及错误信息如下：

代码示例 #

pub async fn new_handler(endpoint: &str) -> tokio::task::JoinHandle<()> {
    // Make master df for this handler
    let mut df = DataFrame::empty().lazy();
    // Make a stream for this handler
    let stream = new_stream(endpoint).await;
    let handle = tokio::spawn(async move {
        stream
            .for_each(|msg| async move {
                match msg {
                    Ok(msg) => {
                        // Parse the json message into a struct
                        let jsonmsg: AggTrade =
                            serde_json::from_str(&msg.to_string()).expect("Failed to parse json");
                        let s0 = Series::new(
                            "price",
                            vec![jsonmsg.price.parse::<f32>().expect("Failed to parse price")],
                        );
                        let s1 = Series::new(
                            "quantity",
                            vec![jsonmsg
                                .quantity
                                .parse::<f32>()
                                .expect("Failed to parse quantity")],
                        );
                        // Create new dataframe from the json data
                        let df2 = DataFrame::new(vec![s0.clone(), s1.clone()]).unwrap().lazy();
                        // append the new data from df2 to the master df
                        df = polars::prelude::concat([df, df2], false, true)
                            .expect("Failed to concat");
                    }
                    Err(e) => {
                        println!("Error: {}", e);
                    }
                }
            })
            .await
    });
    handle
}

错误信息 #

error: captured variable cannot escape `FnMut` closure body
  --> src/websockets.rs:33:29
   |
27 |       let mut df = DataFrame::empty().lazy();
   |           ------ variable defined here
...
33 |               .for_each(|msg| async {
   |  ___________________________-_^
   | |                           |
   | |                           inferred to be a `FnMut` closure
34 | |                 match msg {
35 | |                     Ok(msg) => {
36 | |                         // Parse the json message into a struct
...  |
58 | |                         df = polars::prelude::concat([df.clone(), df2.clone()], false, true)
   | |                                                       -- variable captured here
...  |
86 | |                 }
87 | |             })
   | |_____________^ returns an `async` block that contains a reference to a captured variable, which then escapes the closure body
   |
   = note: `FnMut` closures only have access to their captured variables while they are executing...
   = note: ...therefore, they cannot allow references to captured variables to escape

问题原因 #

for_each的闭包被推断为FnMut类型，它返回的async move块捕获了df，但FnMut闭包不允许捕获的变量引用逃逸出闭包体——异步块会在闭包执行结束后才运行，导致df的生命周期无法满足异步块的需求，从而触发逃逸错误。

解决方法 #

使用线程安全的可变容器Arc<Mutex<...>>包装LazyFrame，让异步块可以安全地共享和修改它。修改后的代码如下：

use std::sync::{Arc, Mutex};
use polars::prelude::*;

pub async fn new_handler(endpoint: &str) -> tokio::task::JoinHandle<()> {
    // 用Arc<Mutex>包装主LazyFrame，支持跨异步安全修改
    let df = Arc::new(Mutex::new(DataFrame::empty().lazy()));
    let stream = new_stream(endpoint).await;
    // 克隆Arc，传递到spawn的任务中
    let df_clone = Arc::clone(&df);
    
    let handle = tokio::spawn(async move {
        stream
            .for_each(|msg| {
                // 再次克隆Arc，传递到内部异步块
                let df_inner = Arc::clone(&df_clone);
                async move {
                    match msg {
                        Ok(msg) => {
                            let jsonmsg: AggTrade = serde_json::from_str(&msg.to_string())
                                .expect("Failed to parse json");
                            
                            let s0 = Series::new(
                                "price",
                                vec![jsonmsg.price.parse::<f32>().expect("Failed to parse price")],
                            );
                            let s1 = Series::new(
                                "quantity",
                                vec![jsonmsg.quantity.parse::<f32>().expect("Failed to parse quantity")],
                            );
                            
                            let df2 = DataFrame::new(vec![s0, s1]).unwrap().lazy();
                            
                            // 锁定Mutex获取可变引用，更新主DataFrame
                            let mut locked_df = df_inner.lock().unwrap();
                            *locked_df = polars::prelude::concat([locked_df.clone(), df2], false, true)
                                .expect("Failed to concat");
                        }
                        Err(e) => {
                            println!("Error: {}", e);
                        }
                    }
                }
            })
            .await
    });
    
    handle
}

关键改动说明 #

• 用Arc<Mutex<LazyFrame>>替代直接的可变变量，实现跨异步任务的安全共享和可变访问
• 在for_each闭包内部再次克隆Arc，确保异步块持有有效的引用
• 通过lock().unwrap()获取Mutex的可变锁，在锁的作用域内完成df的更新操作

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