当调用长于间隔长度时，反应性扩展固定异步调用之间的间隔

Question

这是我的Interval 定义：

m_interval = Observable.Interval(TimeSpan.FromSeconds(5), m_schedulerProvider.EventLoop)
                .ObserveOn(m_schedulerProvider.EventLoop)
                .Select(l => Observable.FromAsync(DoWork))
                .Concat()
                .Subscribe();

在上面的代码中，我在SchedulerProvider 的Interval 和ObserveOn 中输入了IScheduler，这样我就可以更快地进行单元测试（TestScheduler.AdvanceBy）。另外，DoWork 是一个 async 方法。

在我的特定情况下，我希望每 5 秒调用一次 DoWork 函数。这里的问题是我希望 5 秒是 DoWork 结束和另一个开始之间的时间。因此，如果DoWork 的执行时间超过 5 秒，比如说 10 秒，第一次调用将在 5 秒，第二次调用将在 15 秒。

不幸的是，以下测试证明它的行为并非如此：

[Fact]
public void MultiPluginStatusHelperShouldWaitForNextQuery()
{    
    m_queryHelperMock
        .Setup(x => x.CustomQueryAsync())
        .Callback(() => Thread.Sleep(10000))
        .Returns(Task.FromResult(new QueryCompletedEventData()))
        .Verifiable()
    ;

    var multiPluginStatusHelper = m_container.GetInstance<IMultiPluginStatusHelper>();
    multiPluginStatusHelper.MillisecondsInterval = 5000;
    m_testSchedulerProvider.EventLoopScheduler.AdvanceBy(TimeSpan.FromMilliseconds(5000).Ticks);
    m_testSchedulerProvider.EventLoopScheduler.AdvanceBy(TimeSpan.FromMilliseconds(5000).Ticks);

    m_queryHelperMock.Verify(x => x.CustomQueryAsync(), Times.Once);
}

DoWork 调用CustomQueryAsync 并且测试失败，说它被调用了两次。由于.Callback(() => Thread.Sleep(1000)) 强制延迟，它应该只被调用一次。

我在这里做错了什么？

我的实际实现来自this example。

Answer 1

这个问题经常出现，通常是在轮询一些不可观察的数据源时。当我遇到它时，我使用了我不久前写的 RepeatAfterDelay 运算符：

public static IObservable<T> RepeatAfterDelay<T>(this IObservable<T> source, TimeSpan delay, IScheduler scheduler)
{
    var repeatSignal = Observable
        .Empty<T>()
        .Delay(delay, scheduler);

    // when source finishes, wait for the specified
    // delay, then repeat.
    return source.Concat(repeatSignal).Repeat();
}

这就是我使用它的方式：

// do first set of work immediately, and then every 5 seconds do it again
m_interval = Observable
    .FromAsync(DoWork)
    .RepeatAfterDelay(TimeSpan.FromSeconds(5), scheduler)
    .Subscribe();

// wait 5 seconds, then do first set of work, then again every 5 seconds
m_interval = Observable
    .Timer(TimeSpan.FromSeconds(5), scheduler)
    .SelectMany(_ => Observable
        .FromAsync(DoWork)
        .RepeatAfterDelay(TimeSpan.FromSeconds(5), scheduler))
    .Subscribe();

Answer 2

您的问题是您的代码混合了惰性 (Observable) 和非惰性 (Task) 构造。当您的第一个 Task 正在执行时，Interval 将再次触发并在 Select 运算符中创建一个新任务。如果您想避免这种行为，您需要将 Observable 包装到 Defer 块中：

m_interval = Observable.Interval(TimeSpan.FromSeconds(5), m_schedulerProvider.EventLoop)
                .ObserveOn(m_schedulerProvider.EventLoop)
                 //I think `Defer` implicitly wraps Tasks, if not wrap it in `FromAsync` Again
                .Select(l => Observable.Defer(() => DoWork()))
                .Concat()
                .Subscribe();

这样做的结果是每个Observable 只会在订阅时执行延迟的Task，即在前一个完成时。

值得注意的是，如果你的生产者的生产速度比你消耗的快得多，它就会开始堆积，你的记忆就会越来越多。作为替代方案，我建议使用GenerateAsync implementation:

    public static IObservable<TOut> GenerateAsync<TResult, TOut>(
    Func<Task<TResult>> initialState,
    Func<TResult, bool> condition,
    Func<TResult, Task<TResult>> iterate,
    Func<TResult, TimeSpan> timeSelector,
    Func<TResult, TOut> resultSelector,
    IScheduler scheduler = null) 
{
  var s = scheduler ?? Scheduler.Default;

  return Observable.Create<TOut>(async obs => {

    //You have to do your initial time delay here.
    var init = await initialState();
    return s.Schedule(init, timeSelector(init), async (state, recurse) => 
    {
      //Check if we are done
      if (!condition(state))
      {
        obs.OnCompleted();
        return;
      }

      //Process the result
      obs.OnNext(resultSelector(state));

      //Initiate the next request
      state = await iterate(state);

      //Recursively schedule again
      recurse(state, timeSelector(state));

    });
  });
}

GenerateAsync(DoWork /*Initial state*/, 
              _ => true /*Forever*/, 
              _ => DoWork() /*Do your async task*/,
              _ => TimeSpan.FromSeconds(5) /*Delay between events*/, 
              _ => _ /*Any transformations*/, 
              scheduler)
.Subscribe();

上述方法消除了生产者/消费者竞赛的问题，在第一个事件完成之前不安排下一个事件。

Answer 3

虽然@Brandon 的解决方案很好而且很干净，但我发现它阻塞了一个线程来等待延迟计时器。非阻塞替代方案可能类似于：

public static IObservable<T> DelayRepeat<T>(this IObservable<T> source, TimeSpan delay) =>
    source
        .Concat(
            Observable.Create<T>(async observer =>
            {
                await Task.Delay(delay);
                observer.OnCompleted();
            }))
        .Repeat();