Reactive Extensions与FileSystemWatcher

Question

关于FileSystemWatcher的一直困扰我的事情之一是它为文件的单个逻辑更改触发多个事件的方式.我知道它为什么会发生,但我不想要关心 - 我只想重新整理文件一次,而不是连续4-6次.理想情况下,只有在给定文件完成更改时才会触发事件,而不是沿途的每一步.

多年来,我已经提出了各种解决方案来解决这个问题,不同程度的丑陋.我认为Reactive Extensions将是最终解决方案,但是我做得不对,我希望有人可以指出我的错误.

我有一个扩展方法:

public static IObservable<IEvent<FileSystemEventArgs>> GetChanged(this FileSystemWatcher that)
{
    return Observable.FromEvent<FileSystemEventArgs>(that, "Changed");
}

最终,我希望在给定的时间段内为每个文件名获得一个事件 - 这样一行中具有单个文件名的四个事件将减少为一个事件,但如果同一个文件被修改,我不会丢失任何内容时间.BufferWithTime听起来像是理想的解决方案

var bufferedChange = watcher.GetChanged()
    .Select(e => e.EventArgs.FullPath)
    .BufferWithTime(TimeSpan.FromSeconds(1))
    .Where(e => e.Count > 0)
    .Select(e => e.Distinct());

当我订阅这个observable时,对受监视文件的单个更改会连续四次触发我的订阅方法,这相当于失败了.如果我删除了Distinct()呼叫,我看到四个呼叫中的每一个都包含两个相同的事件 - 所以有一些缓冲正在进行.增加TimeSpan传递给BufferWithTime似乎没有任何影响 - 我高达20秒没有任何行为改变.

这是我第一次涉足Rx,所以我可能错过了一些明显的东西.我做错了吗？有更好的方法吗？谢谢你的任何建议......

Answer 1

只是为了热身一个老话题,因为我现在正在努力:

当然,在观看一个文件的上下文中,这个主题可以忽略不计,因为FileSystemWatcher在跟踪大小时通过单个文件的Changed事件每隔~3秒触发一次

_fileSystemWatcher.NotifyFilter = NotifyFilters.Size | ....

但是我们假设FileSystemWatcher会连续发起许多事件(可能会更改/重命名/创建许多文件),其他人会读到这个:

在这种情况下,你不想使用Throttle或BufferWithTime: Throttle有点误导......它禁止任何触发,直到TimeSpan时间过去而没有事件.含义:当你使用类似的东西时,它永远不会激发Throttle(TimeSpan.FromMilliseconds(200)),并且在每个事件之后都有一个<200毫秒的暂停.所以这并不是人们所期望的"限制".当您想要等到用户停止输入内容时,它对用户输入很有用.这对负载限制很不利.

BufferWithTime也不是你想要的:它只是填充了一个时间缓冲.当每个事件的初始加载量很高时很好,例如打开与Web服务的连接.在这种情况下,您希望每隔"时间"秒批处理事件.但不是在负载均衡时,因为事件的数量不会改变.

解决方案是Sample(TimeSpan time)方法:它采用TimeSpan中的最后一个事件,即"真正的"节流.我认为Rx家伙在这种情况下确实搞砸了命名.

Answer 2

您可以使用 group by 来聚合每个文件名的文件系统事件，并将生成的可观察结果与 Throttle 扩展方法一起使用。我使用整数编写了一个小样本，但基本思想是相同的。

var obs = from n in Enumerable.Range(1, 40).ToObservable()
    group n by n / 10 into g
    select new { g.Key, Obs = g.Throttle(TimeSpan.FromMilliseconds(10.0)) } into h
    from x in h.Obs
    select x;
obs.Subscribe(x => Console.WriteLine(x));

输出：

9 
19 
29 
39 
40 

这是每个组 ( n/10) 最后观察到的整数。

Answer 3

我的错。不知何故，我有多个 FileSystemWatchers 监视彼此的文件夹。可观察对象为每个观察者触发一次，但BufferWithTime似乎工作正常。我仍然需要弄清楚为什么我的观察者会触发我认为它们被配置为忽略的文件夹事件，但这与 Rx 或这个问题无关。

事实上，也许我可以解决这个问题，并切换到让单个观察者监视父文件夹，使用 Rx 过滤掉我不感兴趣的文件夹中的事件。