反应式-将可观察间隔与手动触发器相结合

Question

我有Observable.Interval(TimeSpan.FromSeconds(1))和一个订阅者，它在每次发生间隔时都会检查DB中的某些内容。但有时，当我从DB进行检查时，我想立即执行另一次检查（再次呼叫该订户，因为我知道队列中有东西）。

我已经设法通过在订阅者方法中将Interval与while相结合来实现类似的功能：

 Observable              
.Interval(TimeSpan.FromSeconds(1))
.Sample(TimeSpan.FromSeconds(1)) //to avoid multiple 'stacked' intervals 
.Subscribe(RepeatAction);

private void RepeatAction(long _)
{
    bool wasSuccess;
    do
    {
        wasSuccess = CheckingInDB(); //Long operation
    } while (wasSuccess );
}

但是，用纯反应式是否有可能实现这种行为？

Answer 1

是。这是可能的。

首先，你对Rx有一个误解。

如果你运行这个代码：

void Main()
{
    Observable
        .Interval(TimeSpan.FromSeconds(1.0))
        .Sample(TimeSpan.FromSeconds(1.0))
        .Timestamp()
        .Subscribe(RepeatAction);
}
private void RepeatAction(Timestamped<long> _)
{
    Console.WriteLine(_.Timestamp);
    Thread.Sleep(10000);
}

你会得到这样的结果：

2016/05/11 10:37:57 +00:00
2016/05/11 10:38:07 +00:00
2016/05/11 10:38:17 +00:00
2016/05/11 10:38:27 +00:00

您将看到，生成的每个值之间的步长是10秒，而不是1。Interval运算符简单地确保每个值之间的间隔至少是TimeSpan的持续时间，但如果观察者花费更长的时间，则持续时间与每个订阅者花费的时间一样长。它不会对值进行排队。

另一种看待它的方式是.Sample(TimeSpan.FromSeconds(1))什么都不做，因为.Interval(TimeSpan.FromSeconds(1.0))确保值之间的最小间隙已经是1秒。

现在，使用纯Rx运算符来解决问题。试试这个：

var query =
    Observable
        .Interval(TimeSpan.FromSeconds(1.0))
        .Select(_ =>
            Observable
                .While(
                    () => CheckingInDB(),
                    Observable.Return(Unit.Default)))
        .Switch();

这将每秒尝试检查数据库，但一旦达到某个值，它就会快速重复检查，直到没有达到为止。然后等待1秒，然后重试。