使用反应式扩展进行数据库轮询

我必须及时查询数据库以了解遗留系统的状态。我想过将查询包装在Observable，但我不知道正确的方法。

基本上，每 5 秒就会出现相同的查询。但我担心我将不得不面对这些问题：

• 如果查询执行需要 10 秒怎么办？我不想 如果前一个查询仍在处理中，则执行任何新查询。
• 另外，应该有一个超时时间。如果当前查询没有执行 例如，20 秒后，应显示一条信息性消息 记录并应发送新的尝试（相同的查询）。

额外细节：

• 查询只是一个SELECT返回带有状态代码列表的数据集（working, faulted).
• Observable 序列将始终获取从查询接收到的最新数据，类似于 Switch 扩展方法。
• 我想将数据库查询（冗长的操作）包装到一个任务中，但我不确定这是否是最佳选择。

我几乎确定查询应该在另一个线程中执行，但我不知道可观察的应该是什么样子，曾经读过Lee Campbell 的 Rx 简介.

这是使用 Rx 轮询另一个系统的相当经典的案例。大多数人都会用Observable.Interval作为他们的首选运营商，对于大多数人来说这没什么问题。

但是，您对超时和重试有特定要求。在这种情况下，我认为你最好使用运算符组合：

• Observable.Timer允许您在指定时间执行查询
• Timeout识别已溢出的数据库查询
• ToObservable()映射你的Task结果为可观察的序列。
• Retry让您在超时后恢复
• Repeat允许您在成功的数据库查询后继续。这也将保持上一个数据库查询完成与下一个数据库查询开始之间的初始周期/间隙。

这个工作LINQPad代码片段应该显示查询正常工作：

void Main()
{
    var pollingPeriod = TimeSpan.FromSeconds(5);
    var dbQueryTimeout = TimeSpan.FromSeconds(10);

    //You will want to have your Rx query timeout after the expected silence of the timer, and then further maximum silence.
    var rxQueryTimeOut = pollingPeriod + dbQueryTimeout;

    var scheduler = new EventLoopScheduler(ts => new Thread(ts) { Name = "DatabasePoller" });

    var query = Observable.Timer(pollingPeriod, scheduler)
                    .SelectMany(_ => DatabaseQuery().ToObservable())
                    .Timeout(rxQueryTimeOut, Observable.Return("Timeout"), scheduler)
                    .Retry()    //Loop on errors
                    .Repeat();  //Loop on success

    query.StartWith("Seed")
        .TimeInterval(scheduler)    //Just to debug, print the timing gaps.
        .Dump();
}

// Define other methods and classes here
private static int delay = 9;
private static int delayModifier = 1;
public async Task<string> DatabaseQuery()
{
    //Oscillate the delay between 3 and 12 seconds
    delay += delayModifier;
    var timespan = TimeSpan.FromSeconds(delay);
    if (delay < 4 || delay > 11)
        delayModifier *= -1;
    timespan.Dump("delay");
    await Task.Delay(timespan);
    return "Value";
}

结果如下：

Seed 00:00:00.0125407
Timeout 00:00:15.0166379
Timeout 00:00:15.0124480
Timeout 00:00:15.0004520
Timeout 00:00:15.0013296
Timeout 00:00:15.0140864
Value 00:00:14.0251731
Value 00:00:13.0231958
Value 00:00:12.0162236
Value 00:00:11.0138606

样本的关键部分是......

var query = Observable.Timer(TimeSpan.FromSeconds(5), scheduler)
                .SelectMany(_ => DatabaseQuery().ToObservable())
                .Timeout(rxQueryTimeOut, Observable.Return("Timeout"), scheduler)
                .Retry()    //Loop on errors
                .Repeat();  //Loop on success

EDIT:以下是如何得出此解决方案的进一步说明。https://github.com/LeeCampbell/RxCookbook/blob/master/Repository/Polling.md