c# - 多线程文件压缩

Question

我刚刚开始使用线程，我想写一个简单的文件压缩器。它应该创建两个后台线程——一个用于读取，另一个用于写入。第一个应该按小块读取文件并将它们放入队列中，其中 int - 是 chunkId。第二个线程应该使 block 出列并按顺序(使用 chunkId)将它们写入输出流(文件，该线程在开始时创建的文件)。

我做到了。但我不明白为什么在我的程序结束并打开我的 gzip 文件后 - 我看到，我的 block 混合在一起，并且文件没有以前的顺序。

public static class Reader
{
    private static readonly object Locker = new object();

    private const int ChunkSize = 1024*1024;

    private static readonly int MaxThreads;
    private static readonly Queue<KeyValuePair<int, byte[]>> ChunksQueue;
    private static int _chunksComplete;

    static Reader()
    {
        MaxThreads = Environment.ProcessorCount;
        ChunksQueue = new Queue<KeyValuePair<int,byte[]>>(MaxThreads);
    }

    public static void Read(string filename)
    {
        _chunksComplete = 0;

        var tRead = new Thread(Reading) { IsBackground = true };
        var tWrite = new Thread(Writing) { IsBackground = true };

        tRead.Start(filename);
        tWrite.Start(filename);

        tRead.Join();
        tWrite.Join();

        Console.WriteLine("Finished");
    }

    private static void Writing(object threadContext)
    {
        var filename = (string) threadContext;

        using (var s = File.Create(filename + ".gz"))
        {
            while (true)
            {
                var dataPair = DequeueSafe();
                if (dataPair.Value == null)
                    return;

                while (dataPair.Key != _chunksComplete)
                {
                    Thread.Sleep(1);
                }

                Console.WriteLine("write chunk {0}", dataPair.Key);

                using (var gz = new GZipStream(s, CompressionMode.Compress, true))
                {
                    gz.Write(dataPair.Value, 0, dataPair.Value.Length);
                }

                _chunksComplete++;
            }
        }
    }

    private static void Reading(object threadContext)
    {
        var filename = (string) threadContext;

        using (var s = File.OpenRead(filename))
        {
            var counter = 0;
            var buffer = new byte[ChunkSize];
            while (s.Read(buffer, 0, buffer.Length) != 0)
            {
                while (ChunksQueue.Count == MaxThreads)
                {
                    Thread.Sleep(1);
                }

                Console.WriteLine("read chunk {0}", counter);

                var dataPair = new KeyValuePair<int, byte[]>(counter, buffer);

                EnqueueSafe(dataPair);

                counter++;
            }

            EnqueueSafe(new KeyValuePair<int, byte[]>(0, null));
        }
    }

    private static void EnqueueSafe(KeyValuePair<int, byte[]> dataPair)
    {
        lock (ChunksQueue)
        {
            ChunksQueue.Enqueue(dataPair);
        }
    }

    private static KeyValuePair<int, byte[]> DequeueSafe()
    {
        while (true)
        {
            lock (ChunksQueue)
            {
                if (ChunksQueue.Count > 0)
                {
                    return ChunksQueue.Dequeue();
                }
            }

            Thread.Sleep(1);
        }
    } 
}

更新:我只能使用 .NET 3.5

Answer 1

Stream.Read() 返回它消耗的实际字节数。使用它来限制编写器的 block 大小。而且，由于涉及并发读取和写入，您将需要多个缓冲区。尝试将 4096 作为 block 大小。

读者:

var buffer = new byte[ChunkSize]; 
int bytesRead = s.Read(buffer, 0, buffer.Length);

while (bytesRead != 0)
{  
   ...
   var dataPair = new KeyValuePair<int, byte[]>(bytesRead, buffer); 
   buffer = new byte[ChunkSize];
   bytesRead = s.Read(buffer, 0, buffer.Length);
}

作者:

 gz.Write(dataPair.Value, 0, dataPair.Key)

PS:可以通过添加空闲数据缓冲区池而不是每次都分配新缓冲区并使用事件(例如 ManualResetEvent)来通知 队列为空 来提高性能, queue is full 而不是使用 Thread.Sleep()。