Observable.Timer 或 TPL 与 Task.Delay

Question

我有一个要求，在 10 秒的初始延迟后，我需要每 10 分钟执行一次SomeMethod，但需要注意的是，10 分钟计时器应该在SomeMethod 完成后启动。这是粗略的例子：

Start Task 00:00:00
(10 second delay)
SomeMethod executed at 00:00:10 (takes 15 minutes)
(10 minute delay)
SomeMethod executed at 00:25:10 
... and so on.

我知道如何使用 TPL 做到这一点。我可以使用 Task.Delay 启动任务并执行 SomeMethod，然后在每次完成后 (ContinueWith TaskStatus.RanToCompletion)，我创建一个新任务并再次执行 SomeMethod。

我的问题是这可能使用Observable.Timer 吗？比如……

Observable.Timer(TimeSpan.FromSeconds(10), TimeSpan.FromMinutes(10))

此代码的问题在于，如果 SomeMethod 需要 15 分钟，我将运行两个不同的 SomeMethod 实例，这是我不想要的。我希望 10 分钟计时器在 SomeMethod 完成后启动。这可以使用Observable 还是我应该留在TPL？

编辑：忘了说我希望SomeMethod 在它自己的线程中运行。

Answer 1

您应该对使用Observable.Timer 进行更多调查。它几乎就像您想要的一样直接开箱即用。

了解 Rx 的一个重要事项是，它保证您永远不会获得单个订阅的并发执行。虽然 Rx 可能会启用各种多线程场景，但它总是会序列化订阅。

那么，以这个 observable 订阅为例：

Observable
    .Timer(TimeSpan.FromSeconds(10.0), TimeSpan.FromSeconds(2.0))
    .Timestamp()
    .Subscribe(x =>
    {
        Thread.Sleep(5000);
        Console.WriteLine(x.ToString());
    });

我创建了一个 observable，它将等待 10 秒开始发射值，然后尝试每 2 秒发射一个值。

然后我添加了.Timestamp() 以准确记录值的生成时间。

最后，我订阅了一个强制线程休眠 5 秒的观察者。

这是输出的前 4 个值：

0@2015-08-31 10:44:34 +00:00
1@2015-08-31 10:44:39 +00:00
2@2015-08-31 10:44:44 +00:00
3@2015-08-31 10:44:49 +00:00

您会注意到这些值之间的间隔为 5 秒。这非常接近你想要的。 Rx 看到两秒过去了，立即执行下一个值。

但是还有另一个 Rx 运算符可以完全满足您的要求 - .Generate(...)。这是一个非常强大的运算符，可以生成各种可观察的流。

你想这样使用它：

Observable
    .Generate(0, x => true, x => x + 1, x => x,
        x => x == 0 ? TimeSpan.FromSeconds(10.0) : TimeSpan.FromSeconds(2.0))
    .Timestamp()
    .Subscribe(x =>
    {
        Thread.Sleep(5000);
        Console.WriteLine(x.ToString());
    });

在这种情况下，它完全按照您想要的方式工作。这是前十个值：

0@2015-08-31 10:48:27 +00:00
1@2015-08-31 10:48:34 +00:00
2@2015-08-31 10:48:41 +00:00
3@2015-08-31 10:48:48 +00:00
4@2015-08-31 10:48:55 +00:00
5@2015-08-31 10:49:02 +00:00
6@2015-08-31 10:49:09 +00:00
7@2015-08-31 10:49:16 +00:00
8@2015-08-31 10:49:23 +00:00
9@2015-08-31 10:49:30 +00:00

它每 7 秒发射一次。 2 来自 generate 运算符，5 来自观察者。

你显然可以投入你需要的时间。

Answer 2

假设SomeMethod在完成时发出一个OnCompleted事件，我们可以将它写成Observable

//If SomeMethod OnCompleted conforms to .NET Event Pattern
var completedObservable = Observable.FromEventPattern<OnCompletedEventArgs>(
            e => this.OnCompleted += e,
            e => this.OnCompleted += e);

//Subscribe to OnCompleted events
var repeatDisposable = completedObservable.Subscribe(_ => 
                                Observable.Timer(TimeSpan.FromMinutes(10))
                                          .Subscribe(_ => SomeMethod()));

//Start condition
var starterDisposable = Observable.Timer(TimeSpan.FromSeconds(10))
                                  .Subscribe(_ => SomeMethod());