并行压缩30GB大文件：如何维持块的写入顺序？

这确实是多线程处理分块文件时非常典型的痛点——既要利用多线程加速压缩，又要保证输出文件和原始文件的顺序完全一致。咱们一步步拆解解决方案：

先回顾你的现有实现 #

文件读取线程（按32MB分块） #

var inputFileReader = new Thread(() => 
{ 
    var buffer = new byte[_32_MB]; 
    using (var fileStream = File.Open(fileURL, FileMode.Open, FileAccess.Read)) 
    using (var bufferedStream = new BufferedStream(fileStream)) 
    { 
        while (bufferedStream.Read(buffer, 0, _32_MB) != 0) 
        { 
            _queue.Wait(); 
            _queue.Push(buffer); 
        } 
        Console.WriteLine("File reading done."); 
        _applicationIsRunning = false; 
    } 
});

单块压缩逻辑（当前问题：直接返回空数组，且线程不可重用） #

public static byte[] GZip(byte[] bytes) { 
    byte[] res = { }; 
    var compressor = new Thread(() => { 
        using (var memoryStream = new MemoryStream()) 
        using (var gZipStream = new GZipStream(memoryStream, CompressionMode.Compress, false)) { 
            gZipStream.Write(bytes, 0, bytes.Length); 
            res = memoryStream.ToArray(); 
        } 
    }); 
    compressor.Start(); 
    return res; 
}

核心问题分析 #

你提到的两个问题本质是同一个：多线程压缩的异步性破坏了原始块的顺序，而且当前压缩方法存在明显的错误——线程还没执行完就返回了空数组，根本拿不到正确的压缩结果。

解决方案：给块打「顺序标签」+「有序结果容器」+「顺序写入线程」 #

1. 给每个块添加原始顺序索引 #

首先，我们需要给每个从文件中读出的块打上它在原始文件中的顺序标记，这样不管压缩耗时多久，我们都能知道它应该在输出文件中的位置。

先定义一个结构体来存储块的索引和数据：

public struct FileBlock
{
    public int Index; // 块在原始文件中的顺序索引，从0开始
    public byte[] Data; // 块的原始字节数据
}

然后修改读取线程的代码，给每个块分配索引：

// 提前计算总块数，方便后续初始化结果容器
var fileInfo = new FileInfo(fileURL);
var totalBlocks = (int)Math.Ceiling((double)fileInfo.Length / _32_MB);

var inputFileReader = new Thread(() => 
{ 
    var buffer = new byte[_32_MB]; 
    int blockIndex = 0;
    using (var fileStream = File.Open(fileURL, FileMode.Open, FileAccess.Read)) 
    using (var bufferedStream = new BufferedStream(fileStream)) 
    { 
        int bytesRead;
        while ((bytesRead = bufferedStream.Read(buffer, 0, _32_MB)) != 0) 
        {
            // 处理最后一块不足32MB的情况，避免写入多余的空字节
            var blockData = bytesRead == _32_MB ? buffer : buffer.Take(bytesRead).ToArray();
            
            _queue.Wait();
            _queue.Push(new FileBlock { Index = blockIndex++, Data = blockData });
        } 
        Console.WriteLine("File reading done."); 
        _applicationIsRunning = false; 
    } 
});

2. 用线程池实现可重用的压缩线程 #

不要每次压缩都创建新线程，改用.NET的线程池（ThreadPool或Task）来复用线程，既节省资源又提高效率。同时，压缩完成后把结果放到有序的线程安全容器中：

// 用数组存储压缩后的块，索引对应原始块的顺序
private byte[][] _compressedBlocks;
private int _completedBlockCount = 0;
private readonly object _lock = new object();

// 初始化结果容器
_compressedBlocks = new byte[totalBlocks][];

// 压缩方法：接收带索引的块，完成后放到对应位置
public void CompressAndStoreBlock(FileBlock block)
{
    byte[] compressedData;
    using (var memoryStream = new MemoryStream())
    {
        using (var gZipStream = new GZipStream(memoryStream, CompressionMode.Compress, false))
        {
            gZipStream.Write(block.Data, 0, block.Data.Length);
        }
        // 注意：GZipStream关闭后才能正确获取完整的压缩数据
        compressedData = memoryStream.ToArray();
    }

    // 线程安全地更新结果容器
    lock (_lock)
    {
        _compressedBlocks[block.Index] = compressedData;
        _completedBlockCount++;
        // 通知写入线程有新块完成
        Monitor.Pulse(_lock);
    }
}

// 从队列取块并提交到线程池压缩
var blockProcessorThread = new Thread(() =>
{
    while (_applicationIsRunning || _queue.Count > 0)
    {
        if (_queue.TryPop(out FileBlock block))
        {
            // 用线程池复用线程，替代每次创建新Thread
            ThreadPool.QueueUserWorkItem(state => CompressAndStoreBlock((FileBlock)state), block);
        }
        else
        {
            Thread.Sleep(10); // 避免空转消耗CPU
        }
    }
});

3. 单独的顺序写入线程 #

这个线程负责从索引0开始，依次检查并写入连续的已完成块，遇到未完成的块就等待，直到前面的块都完成再继续写入，完美保证输出顺序：

var writeThread = new Thread(() =>
{
    int currentWriteIndex = 0;
    using (var outputStream = File.Open(outputFileUrl, FileMode.Create, FileAccess.Write))
    {
        while (currentWriteIndex < totalBlocks || _applicationIsRunning)
        {
            lock (_lock)
            {
                // 等待当前索引的块完成，或者所有块都处理完毕
                while (currentWriteIndex < totalBlocks && _compressedBlocks[currentWriteIndex] == null)
                {
                    Monitor.Wait(_lock);
                }

                // 写入当前已完成的块
                if (currentWriteIndex < totalBlocks)
                {
                    outputStream.Write(_compressedBlocks[currentWriteIndex], 0, _compressedBlocks[currentWriteIndex].Length);
                    currentWriteIndex++;
                }
            }
        }
        Console.WriteLine("文件写入完成");
    }
});

关键优势总结 #

• 顺序绝对一致：每个块的索引从读取时就固定，写入线程严格按索引顺序写入，完全匹配原始文件的块顺序。
• 线程重用：用线程池处理压缩任务，避免频繁创建销毁线程的开销。
• 内存高效：不需要等待所有块都压缩完再写入，压缩一块就可以写入一块（只要前面的块都完成），内存占用更可控。
• 处理边界情况：正确处理了最后一块不足32MB的问题，避免输出文件出现多余空字节。

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