如何在C#中实现文件格式转换与后续处理的并行联动？

针对你要并行转换文件、转换完成立刻处理，且最终等待全流程结束的需求，用TPL实现生产者-消费者流水线是最稳妥的方案，下面直接上代码和关键说明：

核心逻辑 #

• 生产者：并行执行文件转换，每完成一个就把转换后的临时文件路径放入线程安全队列
• 消费者：从队列中实时取出已完成的文件，并行执行处理逻辑
• 流程收尾：先等所有转换任务完成，再通知消费者停止等待，最后等所有处理任务结束

代码实现 #

1. 基础方法（替换成你的实际逻辑） #

// 模拟文件转换：实际开发中替换成你的格式转换代码
private async Task<string> ConvertFileAsync(string sourceFilePath)
{
    // 模拟转换耗时
    await Task.Delay(1000);
    string tempPath = Path.Combine(Path.GetTempPath(), $"{Path.GetFileNameWithoutExtension(sourceFilePath)}.tmp");
    // 示例：用复制文件模拟转换，实际替换成格式转换逻辑
    File.Copy(sourceFilePath, tempPath, overwrite: true);
    return tempPath;
}

// 模拟转换后文件处理：实际开发中替换成你的业务逻辑
private async Task ProcessConvertedFileAsync(string tempFilePath)
{
    // 模拟处理耗时
    await Task.Delay(500);
    Console.WriteLine($"处理完成：{tempFilePath}");
    // 处理完可删除临时文件
    File.Delete(tempFilePath);
}

2. 生产者-消费者流水线实现 #

public async Task RunFilePipelineAsync(IEnumerable<string> sourceFiles)
{
    // 线程安全的队列，用于传递已转换完成的文件路径
    using var convertedFilesQueue = new BlockingCollection<string>();

    // 启动消费者任务：按CPU核心数创建消费者，控制并发量
    var consumerTasks = Enumerable.Range(0, Environment.ProcessorCount)
        .Select(async _ =>
        {
            // GetConsumingEnumerable会自动等待队列有新数据，直到队列标记完成
            foreach (var tempFilePath in convertedFilesQueue.GetConsumingEnumerable())
            {
                await ProcessConvertedFileAsync(tempFilePath);
            }
        })
        .ToList();

    try
    {
        // 启动生产者任务：并行转换所有源文件
        var producerTasks = sourceFiles.Select(async filePath =>
        {
            string tempPath = await ConvertFileAsync(filePath);
            convertedFilesQueue.Add(tempPath);
        });

        // 等待所有转换任务完成
        await Task.WhenAll(producerTasks);
    }
    finally
    {
        // 所有生产者完成后，标记队列不再添加新数据，消费者会在取完队列后退出
        convertedFilesQueue.CompleteAdding();
    }

    // 等待所有处理任务完成，确保全流程结束
    await Task.WhenAll(consumerTasks);
}

关键细节说明 #

• BlockingCollection：自带线程安全和同步机制，消费者会自动等待队列数据，无需手动写监控逻辑，完美解决你“监控已转换文件”的需求。
• 消费者数量控制：用Environment.ProcessorCount设置消费者数量，避免同时启动过多处理任务导致资源耗尽，也可以根据业务需求手动调整。
• 流程完整性：通过Task.WhenAll分别等待生产者和消费者任务，确保所有转换和处理都完成后才结束整个流程。

简化版方案（无并发控制） #

如果你的文件数量不多，不需要严格控制处理并发数，可以用ContinueWith实现更简洁的逻辑：

public async Task RunSimplePipelineAsync(IEnumerable<string> sourceFiles)
{
    var allTasks = new List<Task>();

    foreach (var filePath in sourceFiles)
    {
        // 每个转换任务完成后，立刻启动对应的处理任务
        var convertTask = ConvertFileAsync(filePath);
        var processTask = convertTask.ContinueWith(async result =>
        {
            await ProcessConvertedFileAsync(result.Result);
        }, TaskContinuationOptions.OnlyOnRanToCompletion);

        allTasks.Add(processTask.Unwrap());
    }

    // 等待所有转换+处理任务完成
    await Task.WhenAll(allTasks);
}

这个写法更简单，但缺点是无法限制处理任务的并发数，文件多的时候可能会占用大量资源，适合小批量文件场景。

总结 #

• 追求并发可控、资源利用率最优：选生产者-消费者模式（BlockingCollection方案）。
• 小批量文件、追求代码简洁：选ContinueWith简化方案。
• 两种方案都基于TPL，符合.NET并行编程的最佳实践，完全满足你“转换完成立刻处理、全流程收尾”的需求。

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