rust - rust : the trait `std::future::Future` is not implemented for `()` 中新线程上的异步循环

Question

我知道这个问题已经被问过很多次了，但我仍然不知道该怎么做(更多内容见下文)。

我正在尝试使用 std::thread::spawn 生成一个新线程，然后在其中运行一个异步循环。

我要运行的异步函数:

#[tokio::main] 
pub async fn pull_tweets(pg_pool2: Arc<PgPool>, config2: Arc<Settings>) {
    let mut scheduler = AsyncScheduler::new();

    scheduler.every(10.seconds()).run(move || {
        let arc_pool = pg_pool2.clone();
        let arc_config = config2.clone();
        async {
            pull_from_main(arc_pool, arc_config).await;
        }
    });

    tokio::spawn(async move {
        loop {
            scheduler.run_pending().await;
            tokio::time::sleep(Duration::from_millis(100)).await;
        }
    });
}

生成一个线程以在其中运行:

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    let handle = thread::spawn(move || async {
        pull_tweets(pg_pool2, config2).await;
    });
}

错误:

error[E0277]: `()` is not a future
  --> src/main.rs:89:9
   |
89 |         pull_tweets(pg_pool2, config2).await;
   |         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ `()` is not a future
   |
   = help: the trait `std::future::Future` is not implemented for `()`
   = note: required by `poll`

The last comment here在概括问题方面做得非常出色。似乎在某些时候需要返回值来实现 IntoFuture 而我没有。我尝试将 Ok(()) 添加到闭包和函数中，但没有成功。

• 添加到闭包中实际上什么都不做
• 添加到异步函数中会给我一个新的但听起来相似的错误:

`Result<(), ()>` is not a future

然后我注意到答案专门谈到extension functions ，我没有使用。 This还谈到了扩展功能。

一些其他的 SO 答案:

• This由缺少 async 引起。
• This和 this是特定于 reqwest 库的。

所以似乎都没有用。有人可以帮助我理解 1) 为什么这里存在错误以及 2) 如何解决它？

注意 1:所有这些都可以通过将 std::thread::spawn 替换为 tokio::task::spawn_blocking 轻松解决。但我有意根据 this article 试验线程生成.

注意 2:关于我想要实现的目标的更广泛背景:我在异步循环中从 Twitter 中提取 150,000 条推文。我想比较 2 种实现:在主运行时上运行与在单独的线程上运行。后者是我挣扎的地方。

注意 3:在我看来，线程和异步任务是两个不同的原语，不能混用。即生成新线程不会影响任务的行为方式，生成新任务只会在现有线程上添加工作。如果这个世界观是错误的(以及我能读到的)，请告诉我。

Answer 1

#[tokio::main] 将您的函数转换为以下内容:

#[tokio::main] 
pub fn pull_tweets(pg_pool2: Arc<PgPool>, config2: Arc<Settings>) {
    let rt = tokio::Runtime::new();
    rt.block_on(async {
        let mut scheduler = AsyncScheduler::new();
        // ...
    });
}

请注意，这是一个同步 函数，它生成一个新的运行时并运行内部 future 直至完成。你不需要 await 它，它是一个独立的运行时，有自己的专用线程池和调度程序:

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    let handle = thread::spawn(move || {
        pull_tweets(pg_pool2, config2);
    });
}

---

请注意，您的原始示例在另一个方面是错误的:

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    let handle = thread::spawn(move || async {
        pull_tweets(pg_pool2, config2).await;
    });
}

即使 pull_tweets 是 一个异步函数，线程也不会做任何事情，因为你所做的只是在 async 中创建另一个 future 堵塞。创建的 future 不会执行，因为 futures lazy(并且该线程中无论如何都没有执行程序上下文)。

我会构造代码以直接在新线程中生成运行时，并从那里调用您想要的任何 async 函数:

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    let handle = thread::spawn(move || {
        let rt = tokio::runtime::Builder::new_multi_thread()
            .enable_all()
            .build()
            .unwrap();
        rt.block_on(async {
            pull_tweets(pg_pool2, config2).await;
        });
    });
}

pub async fn pull_tweets(pg_pool2: Arc<PgPool>, config2: Arc<Settings>) {
    // ...
}