如何从同步线程向Tokio运行时线程传递异步函数？

挑战 #

主线程为同步线程，另有一个运行Tokio Runtime的线程。需要将异步函数从同步线程传递至Tokio线程。

现有代码 #

use std::{future::Future, thread, thread::JoinHandle};
use tokio::{
    sync::mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender}
};

type Job = Box<dyn (FnOnce() -> dyn Future<Output = bool>) + Send>;


pub fn thread_init(receiver: UnboundedReceiver<Job>) {
    let handler = thread::spawn(move || {
            thread_main(receiver);
    });
}

pub fn main() {
   let (sender, receiver) = unbounded_channel::<Job>();
   thread_init(receiver);
}

#[tokio::main(worker_threads = 1)]
async fn thread_main(mut receive: UnboundedReceiver<Job>) {
    loop {
        match receive.try_recv() {
            Err(e) => match e {
                tokio::sync::mpsc::error::TryRecvError::Disconnected => {}
                tokio::sync::mpsc::error::TryRecvError::Empty => {
                    yield_now().await;
                }
            },
            Ok(job) => {
                tokio::spawn(job());
            }
        }
    }
}

报错信息 #

`dyn FnOnce() -> (dyn std::future::Future<Output = bool> + 'static) + std::marker::Send` is not a future
the trait `std::future::Future` is not implemented for `dyn FnOnce() -> (dyn std::future::Future<Output = bool> + 'static) + std::marker::Send`
dyn FnOnce() -> (dyn std::future::Future<Output = bool> + 'static) + std::marker::Send must be a future or must implement `IntoFuture` to be awaited
required for `Box<dyn FnOnce() -> (dyn std::future::Future<Output = bool> + 'static) + std::marker::Send>` to implement `std::future::Future`

`dyn FnOnce() -> (dyn std::future::Future<Output = bool> + 'static) + std::marker::Send` cannot be unpinned
consider using `Box::pin`
required for `Box<dyn FnOnce() -> (dyn std::future::Future<Output = bool> + 'static) + std::marker::Send>` to implement `std::future::Future`

问题 #

如何正确地将异步函数传递给Tokio？

解决方案 #

问题根源 #

1. Job类型定义不符合Tokio调度要求：返回的dyn Future未满足Send + 'static约束，且未固定（Unpin），Tokio需要固定的Future才能调度。
2. thread_main的#[tokio::main]注解误用：该注解用于生成程序入口的同步main函数，子线程中应手动创建Tokio Runtime并运行异步逻辑。
3. yield_now未导入，导致编译错误。

修正步骤 #

1. 调整Job类型：让闭包返回Pin<Box<dyn Future<Output = bool> + Send>>，满足Tokio的调度约束。
2. 手动初始化Tokio Runtime：在子线程中创建Runtime，用block_on运行异步的thread_main函数。
3. 导入yield_now：添加tokio::task::yield_now的导入语句。
4. 正确包装异步任务：发送任务时，用Box::pin包裹异步函数后放入闭包。

修正后的完整代码 #

use std::{future::Future, pin::Pin, thread};
use tokio::{
    sync::mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender},
    task::yield_now,
    runtime::Runtime,
};

// 调整Job类型：返回固定且可发送的Future
type Job = Box<dyn FnOnce() -> Pin<Box<dyn Future<Output = bool> + Send>> + Send>;

pub fn thread_init(receiver: UnboundedReceiver<Job>) {
    let _handler = thread::spawn(move || {
        // 手动创建Tokio Runtime
        let rt = Runtime::new().unwrap();
        rt.block_on(thread_main(receiver));
    });
}

pub fn main() {
    let (sender, receiver) = unbounded_channel::<Job>();
    thread_init(receiver);

    // 示例：发送一个异步任务
    sender.send(Box::new(|| {
        Box::pin(async {
            println!("执行异步任务");
            true
        })
    })).unwrap();

    // 主线程保持运行（示例用，实际业务按需处理）
    thread::sleep(std::time::Duration::from_secs(1));
}

// 去掉#[tokio::main]，改为在子线程的Runtime中运行
async fn thread_main(mut receive: UnboundedReceiver<Job>) {
    loop {
        match receive.try_recv() {
            Err(e) => match e {
                tokio::sync::mpsc::error::TryRecvError::Disconnected => break, // 通道断开时退出循环
                tokio::sync::mpsc::error::TryRecvError::Empty => {
                    yield_now().await;
                }
            },
            Ok(job) => {
                // 现在job()返回的是符合要求的Pin<Box<dyn Future>>，可以直接spawn
                tokio::spawn(job());
            }
        }
    }
}

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