如何在Rust中实现可注册异步回调并在独立线程执行的Scheduler

原代码存在的问题 #

1. 类型不匹配：async move || 生成的闭包返回的是 impl Future，但你定义的回调类型要求返回 Box<dyn Future>，两者无法直接兼容；同时缺少跨线程必须的 Send/Sync 约束，导致无法在Tokio任务中安全传递。
2. 所有权错误：run 方法中直接遍历 shared.lock().await.callbacks 会尝试转移向量元素的所有权，但被Mutex保护的状态不能直接移出；另外调用回调的方式错误，callback.await 应该是 callback().await（回调是返回Future的函数）。
3. 异步运行时缺失：main函数没有启动Tokio异步运行时，无法执行异步代码；同时调用 add_callback 时，scheduler是不可变的，但方法要求 &mut self。

修正后的实现代码 #

use std::{future::Future, pin::Pin, sync::Arc};
use tokio::sync::Mutex;

// 调度器状态：存储异步回调
struct SchedulerState {
    // 回调类型约束：可发送、可同步、静态生命周期，返回Pin包裹的Future
    callbacks: Vec<Box<dyn Fn() -> Pin<Box<dyn Future<Output = ()>>> + Send + Sync + 'static>>,
}

struct Scheduler {
    state: Arc<Mutex<SchedulerState>>,
}

impl Scheduler {
    // 构造函数
    fn new() -> Self {
        Scheduler {
            state: Arc::new(Mutex::new(SchedulerState { callbacks: vec![] })),
        }
    }

    // 添加异步回调：简化用户传入方式，自动处理Future的装箱
    async fn add_callback<F, Fut>(&self, callback: F)
    where
        F: Fn() -> Fut + Send + Sync + 'static,
        Fut: Future<Output = ()> + Send + 'static,
    {
        let mut state = self.state.lock().await;
        // 将用户提供的闭包转换为统一的trait对象
        state.callbacks.push(Box::new(move || Box::pin(callback())));
    }

    // 启动调度器：每个回调在独立Tokio任务中执行（对应多线程运行时的独立线程调度）
    async fn run(&self) {
        let state = self.state.lock().await;
        // 遍历所有回调，为每个回调启动一个独立任务
        for callback in state.callbacks.iter() {
            let cb = callback.clone();
            tokio::spawn(async move {
                cb().await;
            });
        }
    }
}

#[tokio::main]
async fn main() {
    let scheduler = Scheduler::new();

    // 添加第一个异步回调
    scheduler.add_callback(|| async move {
        println!("执行回调1");
        tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
        println!("回调1完成");
    }).await;

    // 添加第二个异步回调
    scheduler.add_callback(|| async move {
        println!("执行回调2");
        tokio::time::sleep(tokio::time::Duration::from_secs(2)).await;
        println!("回调2完成");
    }).await;

    // 启动调度器
    scheduler.run().await;

    // 等待所有任务完成（实际场景可根据需求调整）
    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
}

关键细节说明 #

• 回调类型约束：通过 Send + Sync + 'static 确保回调可以安全跨线程传递，Pin<Box<dyn Future>> 处理异步Future的装箱和固定，满足Tokio任务的要求。
• 简化用户接口：add_callback 使用泛型参数接收任意符合条件的异步闭包，自动将其转换为统一的trait对象，无需用户手动装箱。
• 独立任务执行：run 方法中为每个回调启动一个Tokio任务，在Tokio多线程运行时中，这些任务会被调度到不同的线程执行，满足“独立线程中执行”的需求。
• 异步运行时：使用 #[tokio::main] 宏启动Tokio异步运行时，确保异步代码可以正常执行。

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