.NET Framework下如何用异步I/O高效处理百万级本地文件

核心场景回顾 #

• 需同步源仓库与元数据仓库，涉及约200万文件，单次仅数百个需写入操作
• 当前基于BlockingCollection用多线程消费（Parallel.ForEach），需改造为异步I/O实现
• 固定两阶段处理逻辑：
  1. 阶段1：后台收集现有元数据文件期间，通过File.Exists检查文件存在性，将访问过的文件名记录到processedFiles集合
  2. 阶段2：后台收集完成后，改用existingFiles集合查询文件存在性，移除已访问项，最终删除未被访问的陈旧文件
• 要求阶段切换时不丢失总线中的任何工作项，且必须基于.NET Framework实现优化

关键优化方向 #

1. 用异步队列替代BlockingCollection #

.NET Framework中推荐使用System.Threading.Tasks.Dataflow的BufferBlock<T>作为异步生产者/消费者队列，它原生支持异步等待（ReceiveAsync），比BlockingCollection更适配异步编程模型，避免无意义的线程池资源占用。

2. 优化阶段切换的同步机制 #

替换ManualResetEventSlim为TaskCompletionSource<bool>，贴合异步编程模型，避免线程阻塞：

• 后台收集任务完成时，设置TaskCompletionSource的结果触发切换信号
• 异步消费者在阶段1处理时，同时等待队列新项和切换信号，实现无阻塞的平滑切换

3. 线程安全集合选型优化 #

• processedFiles改用ConcurrentDictionary<string, byte>（用byte占位节省内存），并发查询、添加操作性能远优于普通集合
• existingFiles预先加载为HashSet<string>，阶段2查询时可实现O(1)复杂度，移除操作也更高效

4. 异步文件操作的正确实现 #

• File.Exists无原生异步版本，可将其包装在Task.Run中异步执行（仅在必要时使用，避免过度线程池调度）
• 文件读写改用File.WriteAllTextAsync/File.ReadAllTextAsync（.NET Framework 4.5+支持），真正实现异步I/O，释放线程池资源用于其他任务

优化后的代码示例 #

using System.Collections.Concurrent;
using System.Threading.Tasks.Dataflow;

// 工作项定义
public class MetadataWorkItem
{
    public string FilePath { get; set; }
    public string NewContent { get; set; }
}

public class MetadataSyncService
{
    private readonly BufferBlock<MetadataWorkItem> _asyncBus = new BufferBlock<MetadataWorkItem>();
    private readonly ConcurrentDictionary<string, byte> _processedFiles = new ConcurrentDictionary<string, byte>();
    private readonly TaskCompletionSource<bool> _phaseSwitchTcs = new TaskCompletionSource<bool>();
    private HashSet<string> _existingFiles;

    // 生产者调用此方法投递工作项
    public void PostWorkItem(MetadataWorkItem item) => _asyncBus.Post(item);

    // 启动异步消费者集群
    public async Task StartAsyncConsumers(int consumerCount)
    {
        var consumerTasks = Enumerable.Range(0, consumerCount)
            .Select(_ => ProcessWorkItemsAsync())
            .ToList();

        // 启动后台收集现有文件任务（无阻塞）
        _ = CollectExistingFilesAsync();

        // 等待所有消费者完成处理
        await Task.WhenAll(consumerTasks);

        // 最终清理未被访问的陈旧文件
        DeleteStaleFiles();
    }

    private async Task CollectExistingFilesAsync()
    {
        // 替换为实际的现有文件收集逻辑
        _existingFiles = new HashSet<string>(Directory.EnumerateFiles(@"path\to\metadata\repo", "*", SearchOption.AllDirectories));
        // 完成收集，触发阶段2切换
        _phaseSwitchTcs.SetResult(true);
    }

    private async Task ProcessWorkItemsAsync()
    {
        var phaseSwitchTask = _phaseSwitchTcs.Task;

        while (true)
        {
            // 同时等待队列新项或阶段切换信号，避免阻塞
            var completedTask = await Task.WhenAny(_asyncBus.ReceiveAsync(), phaseSwitchTask);

            if (completedTask == phaseSwitchTask)
            {
                // 进入阶段2，处理队列剩余所有项
                await ProcessRemainingItemsInPhase2Async();
                break;
            }

            // 阶段1处理逻辑
            var workItem = await (Task<MetadataWorkItem>)completedTask;
            await ProcessItemInPhase1Async(workItem);
        }
    }

    private async Task ProcessItemInPhase1Async(MetadataWorkItem item)
    {
        // 标记文件已访问
        _processedFiles.TryAdd(item.FilePath, 0);

        // 异步检查文件存在性
        bool fileExists = await Task.Run(() => File.Exists(item.FilePath));

        if (!fileExists || await FileContentDiffersAsync(item.FilePath, item.NewContent))
        {
            // 异步写入文件
            await File.WriteAllTextAsync(item.FilePath, item.NewContent);
        }
    }

    private async Task ProcessRemainingItemsInPhase2Async()
    {
        while (_asyncBus.TryReceive(out var workItem))
        {
            _processedFiles.TryAdd(workItem.FilePath, 0);

            // 阶段2用内存集合查询，避免重复磁盘IO
            bool fileExists = _existingFiles.Contains(workItem.FilePath);

            if (!fileExists || await FileContentDiffersAsync(workItem.FilePath, workItem.NewContent))
            {
                await File.WriteAllTextAsync(workItem.FilePath, workItem.NewContent);
            }
        }
    }

    private async Task<bool> FileContentDiffersAsync(string filePath, string newContent)
    {
        if (!File.Exists(filePath)) return true;
        string existingContent = await File.ReadAllTextAsync(filePath);
        return !string.Equals(existingContent, newContent, StringComparison.Ordinal);
    }

    private void DeleteStaleFiles()
    {
        // 计算未被访问的陈旧文件
        var staleFiles = _existingFiles.Except(_processedFiles.Keys);
        foreach (var file in staleFiles)
        {
            try
            {
                File.Delete(file);
            }
            catch (IOException)
            {
                // 此处添加异常日志逻辑
            }
        }
    }
}

额外优化点 #

• 批量删除优化：用Parallel.ForEach结合Task.Run实现并行异步删除，提升清理效率（需注意文件系统并发限制）
• 队列限流：通过BufferBlock的BoundedCapacity设置队列上限，避免内存溢出
• 取消支持：添加CancellationToken，允许优雅终止消费流程
• 错误隔离：为每个异步操作添加try/catch，避免单个任务失败导致整个消费者崩溃

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