可观察到的节流

Question

我正在尝试使用响应式扩展来实现事件限制。 我有一个系统，其中可能会针对特定用户或其他实体类型以高频率引发事件。 我需要将事件延迟特定的时间量，一旦超时到期，就会使用最后一个值引发事件。

我做的是这个

 private Subject<int> userBalanceObservable = new Subject<int>();
 userBalanceObservable.Sample(TimeSpan.FromSeconds(sampleSeconds))
            .Subscribe(sample => OnRaiseBalanceEvent(sample));

当事件发生时

userBalanceObservable.OnNext(userId);

编辑

这种方法的问题是事件是为传递给 OnNext 的最后一个值引发的，我真正需要的是为传递给 OnNext 的每个值设置一个延迟。

例如 OnNext(1),OnNext(2),OnNext(3) 我需要延迟调用 1,2,3 而我只得到最后一个值 3。

Answer 1

每次达到采样间隔时，Sample 都会发布最后一个值。如果这是您需要的行为，那很好。看看http://www.introtorx.com/content/v1.0.10621.0/13_TimeShiftedSequences.html#Sample 了解更多信息+其他减缓排放速度的方法。

关于您的问题的新信息：

如果您想在达到某个超时后发出所有值，您可以将这些值分组，直到超时为止（注意：如果您继续添加事件而没有超时，您可能会因为频率而耗尽内存事件排放量）

您可以创建一个缓冲区，直到达到Debounce 超时，请参阅 SO 上的此答案以获取指针：How to implement buffering with timeout in RX

Answer 2

buffering 不工作吗？唯一的“问题”是 OnRaiseBalanceEvent 必须使用 list 而不是一个值，但计算机科学中的所有问题都可以通过另一个级别的间接来解决；）

using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Reactive.Subjects;
using System.Reactive.Linq;

namespace ConsoleApplication1
{
  class Program
  {
    static void Main(string[] args)
    {

      Subject<int> userBalanceObservable = new Subject<int>();
    userBalanceObservable.Buffer(TimeSpan.FromSeconds(2)) //get List of items
                     .Subscribe(sampleList => ProcessSamples(sampleList));

      int cont = 0;

      while (!Console.KeyAvailable)
        {
        userBalanceObservable.OnNext(cont);
        cont++;
        userBalanceObservable.OnNext(cont);
        cont++;
        Thread.Sleep(1000);
      }

    }

    private static void ProcessSamples(IList<int> sampleList)
    {
      Console.WriteLine("[{0}]", string.Join(", ", sampleList.ToArray()));
    }

  }
}