Rx:我如何立即响应,并限制后续请求

Question

我想建立一个可以立即响应事件的Rx订阅,然后忽略在指定的"冷却"时段内发生的后续事件.

开箱即用的Throttle/Buffer方法仅在超时过后响应,这不是我需要的.

下面是一些设置场景的代码,并使用Throttle(这不是我想要的解决方案):

class Program
{
    static Stopwatch sw = new Stopwatch();

    static void Main(string[] args)
    {
        var subject = new Subject<int>();
        var timeout = TimeSpan.FromMilliseconds(500);

        subject
            .Throttle(timeout)
            .Subscribe(DoStuff);

        var factory = new TaskFactory();

        sw.Start();

        factory.StartNew(() =>
        {
            Console.WriteLine("Batch 1 (no delay)");
            subject.OnNext(1);
        });

        factory.StartNewDelayed(1000, () =>
        {
            Console.WriteLine("Batch 2 (1s delay)");
            subject.OnNext(2);
        });

        factory.StartNewDelayed(1300, () =>
        {
            Console.WriteLine("Batch 3 (1.3s delay)");
            subject.OnNext(3);
        });

        factory.StartNewDelayed(1600, () =>
        {
            Console.WriteLine("Batch 4 (1.6s delay)");
            subject.OnNext(4);
        });

        Console.ReadKey();
        sw.Stop();
    }

    private static void DoStuff(int i)
    {
        Console.WriteLine("Handling {0} at {1}ms", i, sw.ElapsedMilliseconds);
    }
}

现在运行此输出的输出是:

批次1(无延迟)

在508ms处理1

批次2(1秒延迟)

批次3(1.3s延迟)

第4批(1.6秒延迟)

处理4在2114ms

请注意,不处理批处理2(这很好!)因为由于节流的性质,我们在请求之间等待500毫秒.批处理3也没有处理(由于它接近批处理4,因此它不太正常,因为它从批处理2发生了超过500毫秒).

我正在寻找的是更像这样的东西:

批次1(无延迟)

在~0ms处理1

批次2(1秒延迟)

处理2~1000秒

批次3(1.3s延迟)

第4批(1.6秒延迟)

在约1600时处理4

请注意,在这种情况下不会处理批处理3(这很好!),因为它发生在批处理2的500毫秒内.

编辑:

这是我使用的"StartNewDelayed"扩展方法的实现:

/// <summary>Creates a Task that will complete after the specified delay.</summary>
/// <param name="factory">The TaskFactory.</param>
/// <param name="millisecondsDelay">The delay after which the Task should transition to RanToCompletion.</param>
/// <returns>A Task that will be completed after the specified duration.</returns>
public static Task StartNewDelayed(
    this TaskFactory factory, int millisecondsDelay)
{
    return StartNewDelayed(factory, millisecondsDelay, CancellationToken.None);
}

/// <summary>Creates a Task that will complete after the specified delay.</summary>
/// <param name="factory">The TaskFactory.</param>
/// <param name="millisecondsDelay">The delay after which the Task should transition to RanToCompletion.</param>
/// <param name="cancellationToken">The cancellation token that can be used to cancel the timed task.</param>
/// <returns>A Task that will be completed after the specified duration and that's cancelable with the specified token.</returns>
public static Task StartNewDelayed(this TaskFactory factory, int millisecondsDelay, CancellationToken cancellationToken)
{
    // Validate arguments
    if (factory == null) throw new ArgumentNullException("factory");
    if (millisecondsDelay < 0) throw new ArgumentOutOfRangeException("millisecondsDelay");

    // Create the timed task
    var tcs = new TaskCompletionSource<object>(factory.CreationOptions);
    var ctr = default(CancellationTokenRegistration);

    // Create the timer but don't start it yet.  If we start it now,
    // it might fire before ctr has been set to the right registration.
    var timer = new Timer(self =>
    {
        // Clean up both the cancellation token and the timer, and try to transition to completed
        ctr.Dispose();
        ((Timer)self).Dispose();
        tcs.TrySetResult(null);
    });

    // Register with the cancellation token.
    if (cancellationToken.CanBeCanceled)
    {
        // When cancellation occurs, cancel the timer and try to transition to cancelled.
        // There could be a race, but it's benign.
        ctr = cancellationToken.Register(() =>
        {
            timer.Dispose();
            tcs.TrySetCanceled();
        });
    }

    if (millisecondsDelay > 0)
    {
        // Start the timer and hand back the task...
        timer.Change(millisecondsDelay, Timeout.Infinite);
    }
    else
    {
        // Just complete the task, and keep execution on the current thread.
        ctr.Dispose();
        tcs.TrySetResult(null);
        timer.Dispose();
    }

    return tcs.Task;
}

Answer 1

这是我的方法.它类似于以前的其他产品,但它并没有遭受过度热心的窗口生产问题.

期望的功能非常类似,Observable.Throttle但是一旦到达就发出合格事件,而不是在节流或采样周期的持续时间内延迟.对于符合条件的事件后的给定持续时间,后续事件将被抑制.

作为可测试的扩展方法给出:

public static class ObservableExtensions
{
    public static IObservable<T> SampleFirst<T>(
        this IObservable<T> source,
        TimeSpan sampleDuration,
        IScheduler scheduler = null)
    {
        scheduler = scheduler ?? Scheduler.Default;
        return source.Publish(ps => 
            ps.Window(() => ps.Delay(sampleDuration,scheduler))
              .SelectMany(x => x.Take(1)));
    }
}

我们的想法是使用一个重载Window来创建非重叠的窗口,使用一个windowClosingSelector使用时间向后移动的源sampleDuration.因此,每个窗口将:(a)由其中的第一个元素关闭,(b)保持打开直到允许新元素.然后我们只需从每个窗口中选择第一个元素.

Rx 1.x版本

Publish上面使用的扩展方法在Rx 1.x中不可用.这是一个替代方案:

public static class ObservableExtensions
{
    public static IObservable<T> SampleFirst<T>(
        this IObservable<T> source,
        TimeSpan sampleDuration,
        IScheduler scheduler = null)
    {
        scheduler = scheduler ?? Scheduler.Default;
        var sourcePub = source.Publish().RefCount();
        return sourcePub.Window(() => sourcePub.Delay(sampleDuration,scheduler))
                        .SelectMany(x => x.Take(1));
    }
}

Answer 2

我经过大量试验和错误后发现的解决方案是用以下内容替换受限制的订阅:

subject
    .Window(() => { return Observable.Interval(timeout); })
    .SelectMany(x => x.Take(1))
    .Subscribe(i => DoStuff(i));

编辑加入保罗的清理工作.

Answer 3

我发布的最初答案有一个缺陷：即该Window方法与 an 一起使用Observable.Interval来表示窗口结束时，会设置无限系列的 500ms 窗口。我真正需要的是一个窗口，该窗口在第一个结果输入主题时开始，并在 500 毫秒后结束。

我的示例数据掩盖了这个问题，因为数据很好地分解到了已经要创建的窗口中。（即0-500ms、501-1000ms、1001-1500ms等）

考虑一下这个时间：

factory.StartNewDelayed(300,() =>
{
    Console.WriteLine("Batch 1 (300ms delay)");
    subject.OnNext(1);
});

factory.StartNewDelayed(700, () =>
{
    Console.WriteLine("Batch 2 (700ms delay)");
    subject.OnNext(2);
});

factory.StartNewDelayed(1300, () =>
{
    Console.WriteLine("Batch 3 (1.3s delay)");
    subject.OnNext(3);
});

factory.StartNewDelayed(1600, () =>
{
    Console.WriteLine("Batch 4 (1.6s delay)");
    subject.OnNext(4);
});

我得到的是：

第 1 批（300 毫秒延迟）

在 356ms 处处理 1

第 2 批（700 毫秒延迟）

750ms 处理 2

第 3 批（1.3 秒延迟）

在 1346ms 处处理 3

第 4 批（1.6 秒延迟）

在 1644ms 处处理 4

这是因为窗口从 0ms、500ms、1000ms 和 1500ms 开始，因此每个窗口都Subject.OnNext很好地适合自己的窗口。

我想要的是：

第 1 批（300 毫秒延迟）

约 300 毫秒处理 1

第 2 批（700 毫秒延迟）

第 3 批（1.3 秒延迟）

在 ~1300ms 处理 3

第 4 批（1.6 秒延迟）

经过一番努力并与同事一起研究了一个小时后，我们使用纯 Rx 和单个局部变量得出了更好的解决方案：

bool isCoolingDown = false;

subject
    .Where(_ => !isCoolingDown)
    .Subscribe(
    i =>
    {
        DoStuff(i);

        isCoolingDown = true;

        Observable
            .Interval(cooldownInterval)
            .Take(1)
            .Subscribe(_ => isCoolingDown = false);
    });

我们的假设是对订阅方法的调用是同步的。如果不是，那么可以引入一个简单的锁。