我如何运行定期并有时按需执行异步任务？

Question

我有一个任务（从 Web 下载内容）定期运行s，在运行s 之间暂停 10 分钟。

如果我的程序发现数据已过时，那么它应该运行立即下载任务除非它已经运行正在。如果下载任务超时，下一个任务应该在超时任务后 10 分钟之后，因此所有未来的任务和暂停都会及时转移。

如何使用 Tokio 执行此操作？

我为运行一系列任务创建了一个库，但尝试使用它来解决我的问题失败了。

mod tasks_with_regular_pauses;

use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use tokio::spawn;
use tokio::sync::mpsc::{channel, Receiver, Sender};
use tokio::sync::Mutex;
use tokio::task::JoinHandle;
use tokio_interruptible_future::{
    interruptible, interruptible_sendable, interruptible_straight, InterruptError,
};

pub type TaskItem = Pin<Box<dyn Future<Output = ()> + Send>>;

/// Execute futures from a stream of futures in order in a Tokio task. Not tested code.
pub struct TaskQueue {
    tx: Sender<TaskItem>,
    rx: Arc<Mutex<Receiver<TaskItem>>>,
}

impl TaskQueue {
    pub fn new() -> Self {
        let (tx, rx) = channel(1);
        Self {
            tx,
            rx: Arc::new(Mutex::new(rx)),
        }
    }
    async fn _task(this: Arc<Mutex<Self>>) {
        // let mut rx = ReceiverStream::new(rx);
        loop {
            let this2 = this.clone();
            let fut = {
                // block to shorten locks lifetime
                let obj = this2.lock().await;
                let rx = obj.rx.clone();
                let mut rx = rx.lock().await;
                rx.recv().await
            };
            if let Some(fut) = fut {
                fut.await;
            } else {
                break;
            }
        }
    }
    pub fn spawn(
        this: Arc<Mutex<Self>>,
        notify_interrupt: async_channel::Receiver<()>,
    ) -> JoinHandle<Result<(), InterruptError>> {
        spawn(interruptible_straight(notify_interrupt, async move {
            Self::_task(this).await;
            Ok(())
        }))
    }
    pub async fn push_task(&self, fut: TaskItem) {
        let _ = self.tx.send(fut).await;
    }
}

Answer 1

我建议使用 select! 而不是可中断的期货来检测循环中的 3 个条件之一：

下载任务完成
数据过时信号
数据过期超时信号

“数据已过时”信号可以使用专用通道传送。

select! 允许等待未来（如下载和超时），并同时从通道读取。有关示例，请参阅 the tutorial。

解决方案草图：

loop {
    // it is time to download
    let download_future = ...; // make your URL request
    let download_result = download_future.await;

    // if the outdated signal is generated while download
    // was in progress, ignore the signal by draining the receiver
    while outdated_data_signal_receiver.try_recv().is_ok() {}

    // send results upstream for processing
    download_results_sender.send(download_result); 

    // wait to re-download
    select! {
        // after a 10 min pause
        _ = sleep(Duration::from_minutes(10)) => break,
        // or by an external signal
        _ = outdated_data_signal_receiver.recv() => break,
    }
}

这个逻辑可以通过 timeout 原语进一步简化：

loop {
    // it is time to download
    let download_future = ...; // make your URL request
    let download_result = download_future.await;

    // if the outdated signal is generated while download
    // was in progress, ignore the signal by draining the receiver
    while outdated_data_signal_receiver.try_recv().is_ok() {}

    // send results upstream for processing
    download_results_sender.send(download_result);

    // re-download by a signal, or timeout (whichever comes first)
    _ = timeout(Duration::from_minutes(10), outdated_data_signal_receiver.recv()).await;
}

我如何运行定期并有时按需执行异步任务？

How do I run an asynchronous task periodically and also sometimes on demand?

rust

scheduled-tasks

rust-tokio

我如何 运行 定期并有时按需执行异步任务？

How do I run an asynchronous task periodically and also sometimes on demand?

rust

scheduled-tasks

rust-tokio

我如何运行定期并有时按需执行异步任务？