如何手动轮询Tokio Runtime以驱动Futures且不阻塞当前线程？

我想在Tokio之上实现一个能集成到其他单线程事件循环的非阻塞当前线程执行器，因此不能使用runtime.block_on()（会阻塞主线程）。但我写的代码里poll_fn完全没运行：

fn main() {
  let runtime = Builder::new_current_thread().enable_all().build().unwrap();

  let mut handle = runtime.spawn(async {
    println!("1");
    tokio::time::sleep(Duration::from_secs(1)).await;
    println!("2");
  });

  // 模拟外部事件循环
  for _ in 0..1000 {
    poll::poll_fn(|ctx| {
      Future::poll(Pin::new(&mut handle), ctx)
    });
  }
}

我已经用smol实现了类似功能（可以正常工作），但希望改用Tokio来利用它的工具集：

fn main() {
  let ex = smol::LocalExecutor::new();

  ex.spawn(async {
    println!("1");
    tokio::time::sleep(Duration::from_secs(1)).await;
    println!("2");
  }).detach();

  // 外部事件循环
  loop { 
    ex.try_tick();
    if ex.is_empty() {
      break;
    }
  }
}

你的Tokio代码无法工作的核心原因是：Tokio的任务依赖其内部调度器驱动（比如定时器、IO事件都需要runtime处理），直接poll任务的JoinHandle无法触发runtime的调度逻辑，自然无法推进任务执行。

要实现类似smol的非阻塞驱动逻辑，你需要主动触发Tokio当前线程runtime的任务调度，而非直接poll单个任务。以下是两种可行的实现方式：

方式一：使用CurrentThread Handle的run_now方法 #

run_now会非阻塞地执行所有当前就绪的任务，直到没有更多可运行的任务为止，完全适配外部事件循环的调用逻辑：

use tokio::runtime::{Builder, current_thread::Handle};
use std::time::Duration;

fn main() {
    let runtime = Builder::new_current_thread().enable_all().build().unwrap();
    let handle = runtime.handle().clone();

    // 在当前线程runtime中提交任务
    handle.spawn(async {
        println!("1");
        tokio::time::sleep(Duration::from_secs(1)).await;
        println!("2");
    });

    // 模拟外部事件循环
    let mut loop_count = 0;
    loop {
        // 非阻塞驱动Tokio调度器，执行就绪任务
        handle.block_on(async {
            Handle::current().run_now().unwrap();
        });

        // 这里可以插入外部事件循环的其他逻辑（比如处理自定义IO事件）

        loop_count += 1;
        // 可以根据实际情况调整循环终止条件，比如检查任务是否完成
        if loop_count > 2000 {
            break;
        }
        // 避免CPU空转，可根据需求调整休眠时间
        std::thread::sleep(Duration::from_millis(1));
    }
}

方式二：使用LocalSet配合poll_until_stalled #

LocalSet用于管理单线程本地任务，poll_until_stalled会一次性处理所有就绪任务，非阻塞返回：

use tokio::{runtime::Builder, task::LocalSet, time::Duration};
use std::future::Future;

fn main() {
    let runtime = Builder::new_current_thread().enable_all().build().unwrap();
    let local_set = LocalSet::new();

    // 向LocalSet提交本地任务
    local_set.spawn_local(async {
        println!("1");
        tokio::time::sleep(Duration::from_secs(1)).await;
        println!("2");
    });

    // 进入runtime上下文
    let mut enter_guard = runtime.enter();

    // 模拟外部事件循环
    for _ in 0..2000 {
        // 非阻塞处理所有就绪任务
        let _ = local_set.poll_until_stalled(&mut enter_guard);

        // 插入外部事件处理逻辑

        // 避免CPU空转
        std::thread::sleep(Duration::from_millis(1));
    }
}

关键说明 #

• Tokio的CurrentThread runtime必须主动驱动调度器才能推进任务，这和smol的try_tick逻辑一致，但Tokio需要显式调用调度方法。
• 不要直接poll单个任务的JoinHandle，任务的执行依赖runtime的调度器处理定时器、IO等事件，单独poll任务无法触发这些逻辑。

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