如何使用 Tokio 产生许多可取消的计时器？

Question

如何使用 Tokio 实现固定数量的计时器，这些计时器会跨线程定期重置和取消？当计时器到期时，将执行回调。

与 Go 的 time.AfterFunc 类似的 API 本质上是我想要的：

package main

import (
    "fmt"
    "time"
)

func main() {
    t := time.AfterFunc(time.Hour, func() {
        // happens every 2 seconds with 1 second delay
        fmt.Println("fired")
    })

    for {
        t.Reset(time.Second)
        time.Sleep(time.Second * 2)
    }
}

我发现的唯一实现（足够）相似 API 的 crate 是 timer，它以一种非常天真的方式实现，通过产生 2 个线程。当计时器经常重置时，这很快就会变得令人望而却步。

显而易见的答案是使用 Tokio，问题是如何优雅地做到这一点。

一个选项是每次更新计时器时生成一个新的绿色线程，并使用原子取消前一个计时器，方法是在这个原子上调节回调的执行，例如这个伪 Rust：

tokio::run({
    // for every timer spawn with a new "cancel" atomic
    tokio::spawn({
        Delay::new(Instant::now() + Duration::from_millis(1000))
            .map_err(|e| panic!("timer failed; err={:?}", e))
            .and_then(|_| {
                if !cancelled.load(Ordering::Acquire) {
                    println!("fired");
                }
                Ok(())
            })
    })
})

问题是我为已经取消的计时器保持状态，可能持续几分钟。另外，看起来也不优雅。

除了tokio::time::Delay，tokio::time::DelayQueue似乎也适用。特别是，通过使用从 "insert".

返回的 Key 引用它们来重置和取消计时器的能力

不清楚如何在多线程应用程序中使用这个库，即：

The return value represents the insertion and is used at an argument to remove and reset. Note that Key is token and is reused once value is removed from the queue either by calling poll after when is reached or by calling remove. At this point, the caller must take care to not use the returned Key again as it may reference a different item in the queue.

这会在通过其键取消计时器的任务与使用来自 DelayQueue 流的计时器事件的任务之间创建竞争条件——导致恐慌或取消不相关的计时器。

Answer 1

您可以将来自 futures-rs 的 Select 组合器与 Tokio 一起使用。它 returns 第一个完成的未来然后 ignores/stops 轮询另一个的结果。

作为第二个未来，我们可以使用来自 oneshot::channel 的接收器来创建一个信号来完成我们的组合器未来。

use futures::sync::oneshot;
use futures::*;
use std::thread;
use std::time::{Duration, Instant};
use tokio::timer::Delay;

fn main() {
    let (interrupter, interrupt_handler) = oneshot::channel::<()>();

    //signal to cancel delayed call
    thread::spawn(move || {
        thread::sleep(Duration::from_millis(500)); //increase this value more than 1000ms to see is delayed call is working or not.
        interrupter
            .send(())
            .expect("Not able to cancel delayed future");
    });

    let delayed = Delay::new(Instant::now() + Duration::from_millis(1000))
        .map_err(|e| panic!("timer failed; err={:?}", e))
        .and_then(|_| {
            println!("Delayed Call Executed!");

            Ok(())
        });

    tokio::run(delayed.select(interrupt_handler).then(|_| Ok(())));
}

Playground

如何使用 Tokio 产生许多可取消的计时器？

How do I spawn many cancellable timers using Tokio?

timer

rust

rust-tokio