Rust中在HashMap复用异步闭包/函数的实现问题

问题分析与修复方案 #

1. 快捷回调函数类型不匹配 #

你的shortcut1和shortcut2是普通异步函数，签名为fn(Interaction) -> impl Future<Output = ()>，但add_shortcut方法的泛型约束要求回调返回Pin<Box<Fut>>，导致类型不兼容。

修复方式：
调整add_shortcut的泛型约束，让它直接兼容普通异步函数，自动完成Future装箱：

• 修改回调约束为C: Fn(Interaction) -> Fut + Send + 'static
• 利用FutureExt::boxed()方法将异步函数的返回值装箱为BoxFuture，适配结构体中存储的闭包类型

2. Axum路由处理函数类型不兼容与共享状态问题 #

start方法中，捕获self的闭包不符合Axum路由处理函数的签名要求；同时Web服务器需要在多线程请求中共享App实例，必须用Arc实现线程安全的状态共享。

修复方式：

• 用Arc<Mutex<HashMap<...>>>包裹shortcuts，确保多线程环境下安全访问回调列表
• 在start_server中将App转为Arc<App>，路由处理函数通过克隆Arc来访问共享状态
• 调整路由处理函数的签名，严格匹配Axum的请求处理规范

修复后的完整代码 #

use axum::{
    routing::post,
    http::StatusCode,
    Json, Router,
};
use std::{net::SocketAddr, future::Future, sync::{Arc, Mutex}, collections::HashMap};
use tracing::{debug, error};
use futures::future::BoxFuture;
use futures::FutureExt;

pub struct Interaction {
    r#type: String,
}

#[derive(Default)]
pub struct App {
    shortcuts: Arc<Mutex<HashMap<String, Box<dyn Fn(Interaction) -> BoxFuture<'static, ()> + Send>>>>,
}

impl App {
    pub fn new() -> Self {
        Self::default()
    }

    async fn handle_interaction(
        &self,
        Json(interaction): Json<Interaction> 
    ) -> StatusCode {
        let t = interaction.r#type.clone();
        debug!("Got shortcut: {}", t);

        let shortcuts = self.shortcuts.lock().unwrap();
        if let Some(closure) = shortcuts.get(&t) {
            // 并发执行回调，忽略JoinHandle结果（可按需处理）
            tokio::task::spawn(closure(interaction));
        } else {
            error!("No shortcut found for type: {}", t);
            return StatusCode::NOT_FOUND;
        }

        StatusCode::OK
    }

    pub fn add_shortcut<C, Fut>(
        mut self, 
        callback_id: &str,
        fun: C,
    ) -> Self 
    where
        Fut: Future<Output = ()> + Send + 'static,
        C: Fn(Interaction) -> Fut + Send + 'static,
    {
        let boxed_fun = Box::new(move |interaction| fun(interaction).boxed());
        self.shortcuts.lock().unwrap().insert(callback_id.to_string(), boxed_fun);
        self
    }

    pub async fn start_server(self, addr: SocketAddr) {
        let app = Arc::new(self);

        // 路由处理函数克隆Arc以访问共享状态
        let router = Router::new()
            .route("/", post(move |json| {
                let app = app.clone();
                async move { app.handle_interaction(json).await }
            }));

        debug!("Listening on {}", addr);

        axum::Server::bind(&addr)
            .serve(router.into_make_service())
            .await
            .unwrap();
    }
}

async fn shortcut1(i: Interaction) {
    println!("Hello, World! Got interaction type: {}", i.r#type);
}

async fn shortcut2(i: Interaction) {
    println!("Hello, Other World! Got interaction type: {}", i.r#type);
}

#[tokio::main]
async fn main() {
    tracing_subscriber::fmt::init();

    let app = App::new()
        .add_shortcut("shortcut1", shortcut1)
        .add_shortcut("shortcut2", shortcut2);

    let addr = SocketAddr::from(([127, 0, 0, 1], 3000));
    app.start_server(addr).await;
}

关键改动说明 #

1. 状态共享：用Arc<Mutex<HashMap<...>>>包裹回调列表，保证多线程请求的安全访问
2. 回调适配：修改add_shortcut泛型约束，自动完成异步函数的装箱操作，无需手动处理Pin和Box
3. 路由兼容：将App转为Arc实例，路由处理函数通过克隆Arc访问共享状态，符合Axum路由签名要求
4. 错误处理：添加回调不存在的分支处理，避免unwrap()引发程序崩溃

内容的提问来源于stack exchange，提问作者Mads Ahlquist Jensen