使用反应式扩展和计时器... "For Dummies"版本

尽管社区在这里，这里，这里和这里提供了耐心的帮助 - 我担心我离真正的理解并不近。我试图将IntroToRx的概念锤炼到我的大脑中，但套用亚瑟王的话："这种新的学习让我感到困惑。再解释一下羊的膀胱和地震。 在我看来，这就是我们不妨谈论的。

我会把它简化为需求（已经改变 - .NET 4.0 是踢球者），而不是可能走上我自己的（可能有缺陷的）实现的错误道路：

我需要使用 .NET Framework 4.0 来编写 Windows 服务。
我需要我的 Windows 服务来调用 Web 服务（WS），并向其传递多个 URL 中的一个，但每次调用只有一个 URL。
我需要能够为每个 URL（1 到 n）定义单独的间隔来调用 WS。例如：对于 URL1，我可能每小时调用一次 WS，对于 URL2，我可能会每四个小时调用一次，等等。
返回特定 URL 的错误，我需要能够停止调用该 URL 的 Web 服务。

就是这样。据我所知，这是一个完美的 Rx 场景。我很确定我想为序列创建Observable.Timer，但是如何在出现错误时禁用计时器？这可能是一个等待发生的CancellationToken（或者我只是完成计时器），但我似乎无法将这些示例并将它们混为一个可靠的、可以理解的解决方案。

Whaddaya 说，S.O.：有人想牵着这个老人的手，用小话解释做这种废话的正确方法吗？如果我没有很好地解释它，或者如果我遗漏了一些东西，我很乐意编辑。

谢谢。

1）我想你想看看Observable.Interval. Observable.Timer是单个值序列，相反，Observable.Interval将继续以给定的间隔生成值。如果对于每个项目/url，此轮询周期是恒定的，那么这可能是您想要的。

2）您可以将IEnumerable（URL）和IObservable（轮询事件）序列混合在一起。

3）错误将终止序列。你可以通过将序列中的值投影（Select）到WebRequest的输出来利用这一点。如果 WebRequest 失败，序列将OnError，但作为 OnError 有效负载传播的除外。

然后慢慢地通过这个;

我们想要获取 URL/周期对的列表，
在给定的时间段内轮询 URL，
如果对某个 URL 的任何调用失败，则停止轮询该 URL（并可能记录它）
如果 Windows 服务已停止，则停止轮询

让我们首先处理要轮询的单个 URL

protected override void OnStart(string [] args)
{
    var resource = new {Url="http://www.bing.com", Period=TimeSpan.FromMinutes(3)};
    var pollingSequence =  Observable.Interval(resource.Period)
                                     .Select(_=>Request(resource.Url));
    var subscription = pollingSequence.Subscribe(
            response=>Log(response),
            ex=>Log(ex)
        ));
    _subscriptions = new CompositeDisposable(subscription);
}
protected override void OnStop()
{
    Dispose();
}
private bool Request(string url)
{
    //TODO:
}
public void Dispose()
{
    _subscriptions.Dispose();
}

为了可视化这一点，我们可以使用"大理石图"。这里每个字符空间代表 1 分钟。OnNext 表示为"o"（大理石）。

    bing    --o--o--o--o-       (every 3rd minute)

为了更准确地说，我们实际上是在事件中获取值（即使我们忽略了该值）

    bing    --0--1--2--3-

然后，我们获取每个事件并将其投影到请求中，因此序列现在转到此（其中"r"表示来自请求的响应）

    bing    --r--r--r--r-

如果任何请求失败，

序列将终止，例如，此处 3 个请求失败，我们显示带有"X"的 OnError

    bing    --r--r--X

现在我们可以将示例扩展到资源列表，即 IEnumerable<IObservable<T>>

protected override void OnStart(string [] args)
{
    var resources = new[] { 
        new {Url="http://www.bing.com", Period=TimeSpan.FromMinutes(3)},
        new {Url="http://www.google.com", Period=TimeSpan.FromMinutes(2)},
        new {Url="http://www.yahoo.com", Period=TimeSpan.FromMinutes(5)},
        new {Url="http://www.stackoverflow.com", Period=TimeSpan.FromMinutes(2)}
    };
    //The same as before, but now we create a polling sequence per resource.
    var pollingSequences = from resource in resources
                        select Observable.Interval(resource.Period)
                                            .Select(_=>Request(resource.Url));
    //Now we cant subscribe to an `IEnumerable<IObservable<T>>` *, 
    //  but we can subscribe to each sequence in it.
    // This turns our queries (IE<IO<T>>) into subscriptions (IE<IDisposable>).
    var subscriptions = pollingSequences.Select(x => 
        x.Subscribe(
            response=>Log(response),
            ex=>Log(ex)
        ));
    _subscriptions = new CompositeDisposable(subscriptions);
}

在这里，我们有一个服务，它将开始调用 Request 方法（供您实现），在为 URL 指定的时间段内传递 URL。如果引发 Request 方法，则不再轮询 Url。如果停止该服务，则会释放订阅，并且不会再进行轮询。

为了可视化这一点，我们可以使用"大理石图"。因此，我们从每个"资源"（空间中的数据）的"行"开始，即。 var resources = new[] { ...};

    bing
    google
    yahoo
    SO

接下来，我们认为每一行也有一个轮询事件序列（时间数据）。从上面的代码（使用 1 个字符 = 1 分钟）

    bing    --1--2--3-      (every 3rd minute)
    google  -1-2-3-4-5      (every 2nd minute)
    yahoo   ----1----2      (every 5 minutes)
    SO      -1-2-3-4-5      (every 2nd minute)

你不能订阅IEnumerable<IObservable<T>>，但有像Concat和Merge这样的Rx方法可以处理这种类型的数据。但是，我们不想使用这些序列，因为它们将序列展平为单个序列，这意味着如果任何失败，所有轮询都将停止。

相关内容

最新更新

热门标签：