async-await - 我如何编写一个异步函数来轮询资源并在资源准备就绪时返回或在几秒钟内重试？

Question

我想编写一个异步函数，它反复轮询网络资源并在准备就绪时返回。我正在使用 future::poll_fn 实现它:

#![feature(async_await)]

/*
[dependencies]
rand = "0.7.0"
futures-preview = "=0.3.0-alpha.18"
*/

use futures::future;
use rand;
use std::task::Poll;

enum ResourceStatus {
    Ready,
    NotReady,
}
use ResourceStatus::*;

// Mocking the function requesting a web resource
fn poll_web_resource() -> ResourceStatus {
    if rand::random::<f32>() < 0.1 {
        Ready
    } else {
        NotReady
    }
}

async fn async_get_resource() {
    // do other works
    future::poll_fn(|ctx| match poll_web_resource() {
        Ready => Poll::Ready(()),
        NotReady => Poll::Pending,
    })
    .await
}

fn main() {
    futures::executor::block_on(async_get_resource());
}

它不起作用，因为当 poll_web_resource() 返回 NotReady 时，任务将永远停止。解决它的一种方法是在每次返回 Pending 时唤醒任务:

future::poll_fn(|ctx| match poll_web_resource() {
    Ready => Poll::Ready(()),
    NotReady => {
        ctx.waker().wake_by_ref();
        Poll::Pending
    }
})
.await

这会产生大量不必要的请求。对于我的用例，理想情况是在资源未准备好时每隔几秒请求一次资源。这是我目前的解决方法:

future::poll_fn(|ctx| match poll_web_resource() {
    Ready => Poll::Ready(()),
    NotReady => {
        let waker = ctx.waker().clone();
        thread::spawn(move || {
            thread::sleep(Duration.from_millis(5000));
            waker.wake();
        });
        Poll::Pending
    }
})
.await

这可行，但它使用一个额外的线程来跟踪超时。我认为应该有更好的方法来做到这一点。我怎样才能更地道地实现相同的目标？

Answer 1

由于您使用的是 async/await 关键字，请编写一个循环，在资源可用时退出，或在资源不可用时等待。可以使用 Tokio 的 Delay 来完成等待。 :

#![feature(async_await)]

use futures; // 0.3.0-alpha.17
use rand; // 0.7.0
use std::time::Duration;
use tokio::timer; // 0.2.0-alpha.1

enum ResourceStatus {
    Ready,
    NotReady,
}
use ResourceStatus::*;

async fn async_get_resource() {
    const SLEEP_TIME: Duration = Duration::from_secs(1);

    loop {
        match poll_web_resource() {
            Ready => return,
            NotReady => {
                // Don't actually use println in production async code.
                println!("Waiting...");
                timer::Delay::new(tokio::clock::now() + SLEEP_TIME).await;
            }
        }
    }
}

fn poll_web_resource() -> ResourceStatus {
    if rand::random::<f32>() < 0.1 {
        Ready
    } else {
        NotReady
    }
}

fn main() {
    let runtime = tokio::runtime::Runtime::new().expect("Unable to create the runtime");
    let _resource = runtime.block_on(async_get_resource());
}