如何将任务添加到在另一个线程上运行的Tokio事件循环中? [英] How do I add tasks to a Tokio event loop that is running on another thread?

问题描述

是否可以在新线程上启动tokio :: Delay以允许主循环继续进行?，但是我仍然不清楚如何实现自己的目标.

I'd like to spin up a Tokio event loop alongside a Rocket server, then add events to this loop later on. I read Is there a way to launch a tokio::Delay on a new thread to allow the main loop to continue?, but it's still not clear to me how to achieve my goal.

推荐答案

为

返回的句柄可用于生成在此运行时上运行的任务，并可被克隆以允许将 Handle 移至其他线程.

这里是一个将事件循环放到一个线程中并让第二个线程生成任务的示例.

Here is an example of spinning up the event loop in one thread and having a second thread spawn tasks on it.

use futures::future; // 0.3.5
use std::{thread, time::Duration};
use tokio::{runtime::Runtime, time}; // 0.2.21

fn main() {
    let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel();
    let (handle_tx, handle_rx) = std::sync::mpsc::channel();

    let tokio_thread = thread::spawn(move || {
        let mut runtime = Runtime::new().expect("Unable to create the runtime");

        eprintln!("Runtime created");

        // Give a handle to the runtime to another thread.
        handle_tx
            .send(runtime.handle().clone())
            .expect("Unable to give runtime handle to another thread");

        // Continue running until notified to shutdown
        runtime.block_on(async {
            shutdown_rx.await.expect("Error on the shutdown channel");
        });

        eprintln!("Runtime finished");
    });

    let another_thread = thread::spawn(move || {
        let handle = handle_rx
            .recv()
            .expect("Could not get a handle to the other thread's runtime");

        eprintln!("Another thread created");

        let task_handles: Vec<_> = (0..10)
            .map(|value| {
                // Run this future in the other thread's runtime
                handle.spawn(async move {
                    eprintln!("Starting task for value {}", value);
                    time::delay_for(Duration::from_secs(2)).await;
                    eprintln!("Finishing task for value {}", value);
                })
            })
            .collect();

        // Finish all pending tasks
        handle.block_on(async move {
            future::join_all(task_handles).await;
        });

        eprintln!("Another thread finished");
    });

    another_thread.join().expect("Another thread panicked");

    shutdown_tx
        .send(())
        .expect("Unable to shutdown runtime thread");

    tokio_thread.join().expect("Tokio thread panicked");
}

Runtime created
Another thread created
Starting task for value 0
Starting task for value 1
Starting task for value 2
Starting task for value 3
Starting task for value 4
Starting task for value 5
Starting task for value 6
Starting task for value 7
Starting task for value 8
Starting task for value 9
Finishing task for value 0
Finishing task for value 5
Finishing task for value 4
Finishing task for value 3
Finishing task for value 9
Finishing task for value 2
Finishing task for value 1
Finishing task for value 7
Finishing task for value 8
Finishing task for value 6
Another thread finished
Runtime finished

这篇文章的修订历史记录中提供了针对Tokio 0.1的解决方案.

The solution for Tokio 0.1 is available in the revision history of this post.

另请参阅:

• 如何从Tokio中的非主线程运行异步任务?

这篇关于如何将任务添加到在另一个线程上运行的Tokio事件循环中?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！