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/