如何使用ReactiveFileSystemWatcher获取文件增量内容

嘿，我懂你的需求——现在你能收到文件变更的通知，但想拿到每次追加的具体内容，对吧？咱们来调整代码实现这个功能，核心思路是跟踪每个文件的最后读取位置，每次变更时从这个位置开始读取新增内容。

修改后的完整代码 #

using RxFileSystemWatcher;
using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Reactive.Linq;

namespace ConsoleApp1
{
    class Program
    {
        // 用来记录每个文件上次读取到的位置（用lock保证线程安全）
        private static readonly Dictionary<string, long> _fileLastReadPositions = new Dictionary<string, long>();

        static void Main(string[] args)
        {
            using (var watcher = new ObservableFileSystemWatcher(c => 
            { 
                c.Path = @"C:\Test"; 
                c.IncludeSubdirectories = true;
                // 按需过滤文件类型，比如只监控txt文件
                c.Filter = "*.txt";
            }))
            {
                watcher.Changed
                    // 解决文件保存时多次触发Changed的问题：合并500ms内的重复通知
                    .Throttle(TimeSpan.FromMilliseconds(500))
                    .Select(changeInfo => 
                    {
                        var filePath = Path.Combine(changeInfo.FullPath);
                        string newContent = string.Empty;

                        // 确保字典操作的线程安全
                        lock (_fileLastReadPositions)
                        {
                            // 第一次监控该文件：记录当前文件长度，后续从这里开始读
                            if (!_fileLastReadPositions.TryGetValue(filePath, out long lastPosition))
                            {
                                using (var fileStream = new FileStream(filePath, FileMode.Open, FileAccess.Read, FileShare.ReadWrite))
                                {
                                    lastPosition = fileStream.Length;
                                    _fileLastReadPositions[filePath] = lastPosition;
                                }
                                return $"首次监控文件 {changeInfo.Name}，当前无新增内容";
                            }

                            try
                            {
                                // 以读写共享模式打开文件（避免文件被锁定时无法读取）
                                using (var fileStream = new FileStream(filePath, FileMode.Open, FileAccess.Read, FileShare.ReadWrite))
                                using (var reader = new StreamReader(fileStream))
                                {
                                    // 跳转到上次读取的位置
                                    fileStream.Seek(lastPosition, SeekOrigin.Begin);
                                    // 读取到文件末尾的所有内容（就是新增的部分）
                                    newContent = reader.ReadToEnd();
                                    // 更新最后读取位置到当前文件末尾
                                    _fileLastReadPositions[filePath] = fileStream.Length;
                                }
                            }
                            catch (IOException ex)
                            {
                                // 处理文件被其他程序锁定的情况
                                return $"读取文件 {changeInfo.Name} 时出错: {ex.Message}";
                            }
                        }

                        return string.IsNullOrEmpty(newContent) 
                            ? $"文件 {changeInfo.Name} 发生变更，但无新增内容" 
                            : $"文件 {changeInfo.Name} 新增内容:\n{newContent}";
                    })
                    .Subscribe(Console.WriteLine);

                watcher.Start();
                Console.WriteLine("文件监控已启动，按回车键退出...");
                Console.ReadLine();
            }
        }
    }
}

关键功能说明 #

• 文件位置跟踪：用_fileLastReadPositions字典保存每个文件的最后读取偏移量，确保每次只读取新增的内容，不会重复读取历史数据。
• Throttle操作符：文件保存时通常会触发2-3次Changed事件，用Throttle(500ms)合并短时间内的重复通知，避免重复处理。
• 共享文件访问：打开文件时使用FileShare.ReadWrite，这样即使文件正在被其他程序写入，我们依然可以读取内容。
• 线程安全：用lock保护字典的读写操作，因为Rx的订阅可能在不同线程执行。
• 异常处理：捕获IOException，处理文件被锁定的场景，防止程序崩溃。

注意事项 #

• 如果文件被覆盖（不是追加），这个逻辑会把覆盖后的内容当作新增内容。如果需要区分追加和覆盖，可以结合ChangeType或者对比文件哈希，但你的场景是持续追加，所以这个问题不大。
• 如果你监控的文件数量很多，可以考虑用ConcurrentDictionary替代普通Dictionary，简化线程安全处理。

内容的提问来源于stack exchange，提问作者user584018