使用 Rx 对我不想在特定时间之前执行的操作进行排队？

Question

摘要：我有一个 Web 应用程序，它在业务对象上执行工作流，有时需要在步骤之间故意等待几秒钟或几分钟。我希望（也许通过 Rx.NET）改进这些工作流程的执行，这样我就不会耗尽线程池并在系统负载较重时使网站无响应。

工作流程的一个非常简化的版本是：

1. 创建一个对象
2. 将数据从系统 A 加载到该对象中
3. 将这些数据发布到系统 B

如果系统 A 关闭，我的应用程序将等待并稍后重试。等待时间按照 GMail 不断增加的重试延迟进行建模：等待 1 秒，随后每次重试加倍（最长为 1 小时）。该应用程序将状态保存到数据库中，因此如果整个应用程序崩溃，当它重新启动时，它将恢复其中断的所有工作流程。

目前（请温柔一点）工作流中的每个步骤都是通过调用 ThreadPool.QueueUserWorkItem 将调用 Thread.Sleep 的方法排队来执行的（如果需要上述重试延迟），然后实际执行该步骤。

如果系统运行良好（没有错误），它可以轻松处理我们向其发送的所有流量，并且线程池可以很好地管理所有这些工作流实例的并行执行。但是，如果系统 B 宕机一段时间，重试次数就会增加，延迟就会增加，很快线程池就会被所有休眠线程填满，导致网站对新请求没有响应。

本质上，我想将所有这些待处理的工作流程放入按（上次执行时间+所需重试延迟）排序的队列中。尽管阅读了很多有关 Rx 的内容并对其感到兴奋，但我从未有机会使用它，但它似乎可能是处理此问题的一种有用方法。如果 Rx 可以神奇地管理在准备触发时吐出这些对象，那么它似乎会

1. 极大地简化和澄清这个逻辑，并
2. 防止浪费地使用大量 99% 的时间都在休眠的线程

。 Rx 新手将不胜感激，即使这只是为了解释为什么这实际上不是 Rx 的一个好的用例。

Answer 1

在这种情况下，我可能会坚持使用您当前的解决方案，因为这一点：

该应用程序会痴迷地将状态保存到数据库中，因此如果整个应用程序崩溃，当它重新启动时，它将恢复所有中断的工作流程。

在启动时通过反序列化“恢复”管道（即 x.Where().Select().Timeout().Bla()）是......棘手的。

如果没有更多信息，很难为您提供更详细的解决方案，如果您不尝试对整个流程进行建模，而只是对事务位进行建模（即从 A 加载，发送到 B），那么它实际上可能与 Rx 配合得很好。

不管怎样，解决线程池耗尽的方法是通过 System.Threading.Timer 类，它告诉线程池在排队新项目之前等待超时。

Answer 2

您肯定必须适应：

public IDisposable StartProcess<T>(Action<T> load, Action<T> post) where T : new()
{
    return StartProcess(TimeSpan.FromSeconds(1), new T())
                .Do(load)
                .Subscribe(post);
}

private IObservable<long> StartProcess<T>(TimeSpan span, T obj) where T : new()
{
    Observable
        .Interval(span)
        .OnErrorResumeNext(Observable.Defer(() => StartProcess(IncreaseSpan(span), obj)))
        .Concat(Observable.Defer(() => StartProcess(TimeSpan.FromSeconds(1), new T())));
}

private TimeSpan IncreaseSpan(TimeSpan span)
{
    return TimeSpan.FromSeconds(span.TotalSeconds < 1800? span.TotalSeconds * 2 : 3600);
}

现在我宁愿让 load 实例化并填充对象，而不是显式地执行它，因为函数式编程不鼓励可变性，并且您可能希望 load 实际上转到数据库并恢复就像你提到的那样。

我不确定您是否想保留状态对象，以防对 post 或 load 的调用崩溃，并且您需要进行调整，因为目前，它将保留说明 load 或 post 是否崩溃，如果 post 崩溃，将再次调用 load 而不提供新状态（这可能绝对不会）成为你想做的事。

我没有测试代码，但 Rx 适合你想做的事情。

Answer 3

查看 Rx 论坛上的这篇文章。对于您想要解决的问题来说，非常方便的运算符： http://social.msdn.microsoft.com/Forums/en-US/rx/thread/af43b14e-fb00-42d4-8fb1-5c45862f7796/

Rx 是处理像这样的问题（尤其是），因为您可以拥有异步函数/可观察量并将通用运算符（例如所描述的重试运算符）应用于它们。