public interface Event
 {
      Guid identifier;
      Timestamp ts;
 }

我们正在考虑使用 Reactive Extensions 来重写我的金融公司的一个问题。

前提是我们获得由 Guid(股票代码 + 嵌入其中的唯一性熵)、时间戳和值字段标识的事件。这些发生的频率很高,我们不能对这些对象采取行动,直到“至少”在 X 秒(10 秒)之后,我们必须对它们采取行动,并将它们从系统中删除。

把它想象成两个窗口,一个“10 秒”的初始窗口(例如 T0 到 T10),在那里我们识别所有独特的事件(基本上,按 guid 分组),然后我们查看下一个“10 秒”,“次窗口”(T10-T20),以确保我们正在实现“至少”10 秒的政策。从“初始窗口”中,我们删除所有事件(因为我们已经考虑了它们),然后从“次要窗口”中,我们删除了在“初始窗口”中发生的事件。我们继续移动 10 秒滑动窗口,所以现在我们正在查看窗口 T20-T30,重复并冲洗。

我怎么能在 Rx 中实现它,因为它似乎是要走的路。

最佳答案

如果您可以依赖服务器时钟和消息中的时间戳(即,我们处于“现实生活”模式),那么 您将经历 10 秒的滑动延迟,而不是跳跃的 10 秒窗口,那么你可以将事件延迟 10 秒:

var events = new Subject<Event>();
var delayedEvents = events.Delay(TimeSpan.FromSeconds(10));

检查唯一事件等只是将它们添加到某种集合中的问题:
var guidSet = new HashSet<Guid>();
delayedEvents.Do(e => guidSet.Add(e.identifier));

如果您的问题是您必须等待 10 秒,然后一次处理最后 10 秒,那么您只想缓冲 10 秒:
var bufferedEvents = events.Buffer(TimeSpan.FromSeconds(10));
bufferedEvents.Do(es => { foreach (var e in es) guidSet.Add(e.identifier); });

我没有展示滑动 10 秒窗口的示例,因为我无法想象这就是您想要的(事件被处理不止一次)。

现在我们开始认真了。假设您不想依赖挂墙时间,而是想使用事件中的时间来驱动您​​的逻辑。假设事件被重新定义为:
 public class Event
 {
      public Guid identifier;
      public DateTime ts;
 }

创建历史调度程序并从原始事件中获取调度事件:
var scheduler = new HistoricalScheduler();
var driveSchedule = events.Subscribe(e => scheduler.AdvanceTo(e.ts));
var target = events.SelectMany(e => Observable.Timer(e.ts, scheduler).Select(_ => e));

现在你可以简单地在 target 上使用常规的 Rx 组合器而不是 event ,并且只需通过调度程序以便适本地触发它们,例如:
var bufferedEvents = target.Buffer(TimeSpan.FromSeconds(10), scheduler);

这是一个简单的测试。创建一百个事件,每个事件“几乎”相隔 30 秒,但每秒实时触发:
var now = DateTime.Now;
var test = Enumerable.Range(0,99).Select(i =>
    Scheduler.ThreadPool.Schedule(
        TimeSpan.FromSeconds(i),
        () => events.OnNext(new Event() {
            identifier = Guid.NewGuid(),
            ts = now.AddSeconds(i * 30)
        })
    )
).ToList();

订阅它并请求 60 秒的缓冲事件 - 实际上每 2 个“真实”秒(60 虚拟秒)接收 2 个事件:
target.Select(e => String.Format("{0} {1}", e.identifier, e.ts.ToString()))
      .Buffer(TimeSpan.FromSeconds(60), scheduler)
      .Select(es => String.Join(" - ", es))
      .DumpLive();

关于c# - 使用响应式(Reactive)扩展(rx)枚举及时发生的事件,我们在Stack Overflow上找到一个类似的问题:https://stackoverflow.com/questions/10455395/

10-14 15:41
查看更多