【问题标题】:Am I doing something wrong or is it not possible to extract a zip file in parallel?我做错了什么还是无法并行提取 zip 文件?
【发布时间】:2013-04-24 18:29:40
【问题描述】:

我创建它是为了测试并行提取:

    public static async Task ExtractToDirectoryAsync(this FileInfo file, DirectoryInfo folder)
    {

        ActionBlock<ZipArchiveEntry> block = new ActionBlock<ZipArchiveEntry>((entry) =>
        {
            var path = Path.Combine(folder.FullName, entry.FullName);

            Directory.CreateDirectory(Path.GetDirectoryName(path));
            entry.ExtractToFile(path);

        }, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 2 });

        using (var archive = ZipFile.OpenRead(file.FullName))
        {
            foreach (var entry in archive.Entries.Where(e => e.Name != string.Empty))
            {
                block.Post(entry);
            }
            block.Complete();
            await block.Completion;
        }

    }

以及下面的单元测试进行测试:

    [TestMethod]
    public async Task ExtractTestAsync()
    {
        if (Resources.LocalExtractFolder.Exists)
            Resources.LocalExtractFolder.Delete(true);
        //  Resources.LocalExtractFolder.Create();
        await Resources.WebsiteZip.ExtractToDirectoryAsync(Resources.LocalExtractFolder);
    }

在 MaxDegreeOfParallelism = 1 时,一切正常,但在 2 时则不行。

Test Name:  ExtractTestAsync
Test FullName:  Composite.Azure.Tests.ZipFileTests.ExtractTestAsync
Test Source:    c:\Development\C1\local\CompositeC1\Composite.Azure.Tests\ZipFileTests.cs : line 21
Test Outcome:   Failed
Test Duration:  0:00:02.4138753

Result Message: 
Test method Composite.Azure.Tests.ZipFileTests.ExtractTestAsync threw exception: 
System.IO.InvalidDataException: Unknown block type. Stream might be corrupted.
Result StackTrace:  
at System.IO.Compression.Inflater.Decode()
   at System.IO.Compression.Inflater.Inflate(Byte[] bytes, Int32 offset, Int32 length)
   at System.IO.Compression.DeflateStream.Read(Byte[] array, Int32 offset, Int32 count)
   at System.IO.Stream.InternalCopyTo(Stream destination, Int32 bufferSize)
   at System.IO.Stream.CopyTo(Stream destination)
   at System.IO.Compression.ZipFileExtensions.ExtractToFile(ZipArchiveEntry source, String destinationFileName, Boolean overwrite)
   at System.IO.Compression.ZipFileExtensions.ExtractToFile(ZipArchiveEntry source, String destinationFileName)
   at Composite.Azure.Storage.Compression.ZipArchiveExtensions.<>c__DisplayClass6.<ExtractToDirectoryAsync>b__3(ZipArchiveEntry entry) in c:\Development\C1\local\CompositeC1\Composite.Azure.Storage\Compression\ZipArchiveExtensions.cs:line 37
   at System.Threading.Tasks.Dataflow.ActionBlock`1.ProcessMessage(Action`1 action, KeyValuePair`2 messageWithId)
   at System.Threading.Tasks.Dataflow.ActionBlock`1.<>c__DisplayClass5.<.ctor>b__0(KeyValuePair`2 messageWithId)
   at System.Threading.Tasks.Dataflow.Internal.TargetCore`1.ProcessMessagesLoopCore()
--- End of stack trace from previous location where exception was thrown ---
   at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
   at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
   at System.Runtime.CompilerServices.TaskAwaiter.GetResult()
   at Composite.Azure.Storage.Compression.ZipArchiveExtensions.<ExtractToDirectoryAsync>d__8.MoveNext() in c:\Development\C1\local\CompositeC1\Composite.Azure.Storage\Compression\ZipArchiveExtensions.cs:line 48
--- End of stack trace from previous location where exception was thrown ---
   at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
   at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
   at System.Runtime.CompilerServices.TaskAwaiter.GetResult()
   at Composite.Azure.Tests.ZipFileTests.<ExtractTestAsync>d__2.MoveNext() in c:\Development\C1\local\CompositeC1\Composite.Azure.Tests\ZipFileTests.cs:line 25
--- End of stack trace from previous location where exception was thrown ---
   at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
   at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
   at System.Runtime.CompilerServices.TaskAwaiter.GetResult()

更新 2

这是我自己的并行操作,它也不起作用 :) 记得在 continueWith 中处理异常。

public static void ExtractToDirectorySemaphore(this FileInfo file, DirectoryInfo folder)
        {

            int MaxDegreeOfParallelism = 2;
            using (var archive = ZipFile.OpenRead(file.FullName))
            {

                var semaphore = new Semaphore(MaxDegreeOfParallelism, MaxDegreeOfParallelism);

                foreach (var entry in archive.Entries.Where(e => e.Name != string.Empty))
                {
                    semaphore.WaitOne();

                    var task = Task.Run(() =>
                    {
                        var path = Path.Combine(folder.FullName, entry.FullName);

                        Directory.CreateDirectory(Path.GetDirectoryName(path));
                        entry.ExtractToFile(path);
                    });
                    task.ContinueWith(handle =>
                    {
                        try
                        {
                            //do any cleanup/post processing
                        }
                        finally
                        {
                            // Release the semaphore so the next thing can be processed
                            semaphore.Release();
                        }
                    });
                }
                while(MaxDegreeOfParallelism-->0)
                    semaphore.WaitOne(); //Wait here until the last task completes.


            }

        }

这是异步版本:

public static Task ExtractToDirectorySemaphoreAsync(this FileInfo file, DirectoryInfo folder)
        {
            return Task.Factory.StartNew(() =>
            {
                int MaxDegreeOfParallelism = 50;
                using (var archive = ZipFile.OpenRead(file.FullName))
                {

                    var semaphore = new Semaphore(MaxDegreeOfParallelism, MaxDegreeOfParallelism);

                    foreach (var entry in archive.Entries.Where(e => e.Name != string.Empty))
                    {
                        semaphore.WaitOne();

                        var task = Task.Run(() =>
                        {
                            var path = Path.Combine(folder.FullName, entry.FullName);

                            Directory.CreateDirectory(Path.GetDirectoryName(path));
                            entry.ExtractToFile(path);
                        });
                        task.ContinueWith(handle =>
                        {
                            try
                            {
                                //do any cleanup/post processing
                            }
                            finally
                            {
                                // Release the semaphore so the next thing can be processed
                                semaphore.Release();
                            }
                        },TaskContinuationOptions.AttachedToParent); // the outher task will wait for all.
                    }

                }
            });
        }

更新 3

handle.Exception 中抛出以下异常。

{"Block length does not match with its complement."}  
[0] = {"A local file header is corrupt."}

必须确定 ZipFile 是否是线程安全的。

【问题讨论】:

  • 并行版本比非并行版本快多少?
  • 嗯,它是用于上传到 azure 存储,其中将添加更多内容,然后只是提取文件。但是在一个简单的提取到文件夹中,它从 10 秒到 5 秒发泄了 2 个并发任务。 n 个并发任务没有进一步的好处。但我的 zip 文件是 1500 个文件,每个文件 0-50kb 和一些大文件。由于不受上下文切换的限制,较大文件的速度会更快。
  • 我怀疑存在竞争条件。如果 zip 中的文件非常大……比如兆字节或更多,看看你的并行(非异步)版本是否有效会很有趣。你能测试一下吗?创建一个包含几个 1 兆字节文件的 zip?
  • ZipArchiveEntry 具有对ZipFile 的引用(因此它可以提取或删除),因此当您调用ExtractToFile 时,您最终会在多个上同时使用原始ZipFile线程,这是不允许的。
  • @RaymondChen 所说的:但作为一种解决方案,您可以复制输入流(意味着一些 byte[]),创建单独的 ZipFiles,然后并行提取它们(显然您需要手动分区文件条目):)

标签: c# .net asynchronous tpl-dataflow


【解决方案1】:

Disclamer:它只是一个概念证明。

在代码中的示例中将 ZipFile.OpenRead 替换为 ParallelZipFile.OpenRead,所有 4 个单元测试都通过了。

   public class ParallelZipFile
    {
        public static ParallelZipArchive OpenRead(string path)
        {

            return new ParallelZipArchive(ZipFile.OpenRead(path),path);
        }
    }
    public class ParallelZipArchive : IDisposable
    {
        internal ZipArchive _archive;
        internal string _path;
        internal ConcurrentQueue<ZipArchive> FreeReaders = new ConcurrentQueue<ZipArchive>();

        public ParallelZipArchive(ZipArchive zip,string path)
        {
            _path = path;
            _archive = zip;
            FreeReaders.Enqueue(zip);
        }

        public ReadOnlyCollection<ParallelZipArchiveEntry> Entries
        {
            get
            {
                var list = new List<ParallelZipArchiveEntry>(_archive.Entries.Count);
                int i = 0;
                foreach (var entry in _archive.Entries)
                    list.Add(new ParallelZipArchiveEntry(i++, entry, this));

                return  new ReadOnlyCollection<ParallelZipArchiveEntry>(list);
            }
        }


        public void Dispose()
        {
            foreach (var archive in FreeReaders)
                archive.Dispose();
        }
    }
    public class ParallelZipArchiveEntry
    {
        private ParallelZipArchive _parent;
        private int _entry;
        public string Name { get; set; }
        public string FullName { get; set; }

        public ParallelZipArchiveEntry(int entryNr, ZipArchiveEntry entry, ParallelZipArchive parent)
        {
            _entry = entryNr;
            _parent = parent;
            Name = entry.Name;
            FullName = entry.FullName;
        }

        public void ExtractToFile(string path)
        {
            ZipArchive value;
            Trace.TraceInformation(string.Format("Number of readers: {0}", _parent.FreeReaders.Count));

            if (!_parent.FreeReaders.TryDequeue(out value))
                value = ZipFile.OpenRead(_parent._path);

            value.Entries.Skip(_entry).First().ExtractToFile(path);



            _parent.FreeReaders.Enqueue(value);
        }
    }

单元测试

[TestClass]
    public class ZipFileTests
    {
        [ClassInitialize()]
        public static void PreInitialize(TestContext context)
        {
            if (Resources.LocalExtractFolderTruth.Exists)
                Resources.LocalExtractFolderTruth.Delete(true);

            ZipFile.ExtractToDirectory(Resources.WebsiteZip.FullName, Resources.LocalExtractFolderTruth.FullName);
        }

        [TestInitialize()]
        public void InitializeTests()
        {
            if (Resources.LocalExtractFolder.Exists)
                Resources.LocalExtractFolder.Delete(true);

        }

        [TestMethod]
        public void ExtractTest()
        {

            Resources.WebsiteZip.ExtractToDirectory(Resources.LocalExtractFolder);

            Assert.IsTrue(Helpers.DirectoryTools.CompareDirectories(
                Resources.LocalExtractFolderTruth, Resources.LocalExtractFolder));

        }
        [TestMethod]
        public async Task ExtractAsyncTest()
        {

            await Resources.WebsiteZip.ExtractToDirectoryAsync(Resources.LocalExtractFolder);

            Assert.IsTrue(Helpers.DirectoryTools.CompareDirectories(
               Resources.LocalExtractFolderTruth, Resources.LocalExtractFolder));
        }
        [TestMethod]
        public void ExtractSemaphoreTest()
        {

            Resources.WebsiteZip.ExtractToDirectorySemaphore(Resources.LocalExtractFolder);
            Assert.IsTrue(Helpers.DirectoryTools.CompareDirectories(
               Resources.LocalExtractFolderTruth, Resources.LocalExtractFolder));
        }
        [TestMethod]
        public async Task ExtractSemaphoreAsyncTest()
        {

            await Resources.WebsiteZip.ExtractToDirectorySemaphoreAsync(Resources.LocalExtractFolder);
            Assert.IsTrue(Helpers.DirectoryTools.CompareDirectories(
               Resources.LocalExtractFolderTruth, Resources.LocalExtractFolder));
        }

    }

【讨论】:

  • .NET 框架的内置 ZipArchive 类型是否有类似的线程安全方法?
  • 此解决方案不会受益于 ZipArchive 保存的缓存条目列表。如果 zip 文件有很多小文件,这种方案比顺序读取要慢。
【解决方案2】:

最近从事相同的任务,这是我的结果:

我在我的 Xeon 1socket、8cores、16cpu 上使用 DotNetZip.reduced (Ionic.Zip.Reduced.dll v1.9.1.8); 32GB 内存; SSD 驱动器

压缩文件 |包装尺寸 |解压文件 |未包装尺寸

  • 小文件1 | 778 MB | 4,926 个文件 | 1.4 GB
  • 大文件2 | 6 GB | 29,557 个文件 | 10.0 GB

我有 5 种方法:首先在一个线程中执行所有操作,另外 4 个使用 PLINQ 和 TPL Parallel 类。

获胜者是 V4 和 V5,它们的工作速度比 V1 快 6 倍。以下是详细结果和代码。

  • V1 使用 ExtractAll
  • V2 并行提取条目(非线程安全)
  • V3 通过为每个条目打开新文件句柄来并行提取条目
  • V4 仅使用 N+1 个文件句柄并行提取条目
  • V5 最终版

性能结果表 压缩文件 | V1,秒 | V2,秒 | V3,秒 | V4,秒 | V5,秒

  • 小文件1 | 32 |例外 | 8 | 8 | 5
  • 大文件2 | 200 |例外 | 2000 | 35 | 30

小文件处理

V1 处理大文件

V4 处理大文件

V5 处理大文件

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.Linq;
using System.Reflection;
using System.Threading;
using System.Threading.Tasks;
using Ionic.Zip;
using Ionic.Zlib;

namespace A1
{
    public static class Program
    {
        static void Main(string[] args)
        {
            Console.ReadKey();

            CancellationToken cancellationToken = CancellationToken.None;

            string path = @"e:\1\";
            string zf1 = Path.Combine(path, "1.zip");
            string zf2 = Path.Combine(path, "2.zip");
            Stopwatch sw = new Stopwatch();

            List<string> zipFiles = new List<string>
            {
                zf1,
                zf2,
            };
            List<Action<string, string, CancellationToken>> methods = new List<Action<string, string, CancellationToken>>
            {
                ExtractAllFilesFromZipFileV1,
                //ExtractAllFilesFromZipFileV2,
                //ExtractAllFilesFromZipFileV3,
                ExtractAllFilesFromZipFileV4,
                ExtractAllFilesFromZipFileV5,
                ExtractAllFilesFromZipFileV4,
                ExtractAllFilesFromZipFileV5,
                ExtractAllFilesFromZipFileV4,
                ExtractAllFilesFromZipFileV5,
            };

            zipFiles.Reverse();
            methods.Reverse();

            zipFiles.ForEach(f => methods.ForEach(m =>
            {
                string fileName = Path.GetFileName(f);
                string targetDirectory = path + Guid.NewGuid().ToString("N");
                sw.Restart();
                // Unzip
                try
                {
                    m(f, targetDirectory, cancellationToken);
                }
                catch (Exception ex)
                {
                    Console.WriteLine(ex.Message);
                }
                sw.Stop();
                Console.WriteLine("{0} processed by {1} in {2} seconds", fileName, m.GetMethodInfo().Name, sw.Elapsed.TotalSeconds.ToString("F3"));
                Thread.Sleep(5 * 1000);
                Directory.Delete(targetDirectory, true);
                Thread.Sleep(5 * 1000);
            }));
        }

        private static void ExtractAllFilesFromZipFileV1(string zipFileName, string targetDirectory, CancellationToken cancellationToken)
        {
            using (ZipFile zipFile = new ZipFile(zipFileName))
            {
                zipFile.ExtractAll(targetDirectory);
            }
        }

        private static void ExtractAllFilesFromZipFileV2(string zipFileName, string targetDirectory, CancellationToken cancellationToken)
        {
            using (ZipFile zipFile = new ZipFile(zipFileName))
            {
                zipFile.Entries
                    .AsParallel()
                    .ForAll(v =>
                    {
                        v.Extract(targetDirectory);
                    });
            }
        }


        private static void ExtractAllFilesFromZipFileV3(string zipFileName, string targetDirectory, CancellationToken cancellationToken)
        {
            using (ZipFile zipFile = new ZipFile(zipFileName))
            {
                int count = zipFile.Entries.Count;

                Enumerable.Range(0, count)
                    .AsParallel()
                    .ForAll(v =>
                    {
                        cancellationToken.ThrowIfCancellationRequested();

                        using (ZipFile zf = new ZipFile(zipFileName))
                        {
                            // Get the right entry to extract
                            zf.Entries
                                .Skip(v)
                                .First()
                                .Extract(targetDirectory);
                        }
                    });
            }
        }

        private static void ExtractAllFilesFromZipFileV4(string zipFileName, string targetDirectory, CancellationToken cancellationToken)
        {
            using (ZipFile zipFile = new ZipFile(zipFileName))
            {
                // Get count of files, files and keep the lock on the file
                int count = zipFile.Entries.Count();

                // Cache instances of ZipFile used by threads
                // Make sure that we have only open zip file not more than N times, where N is maxDop.
                ConcurrentDictionary<int, ZipFile> dictionary = new ConcurrentDictionary<int, ZipFile>();

                try
                {
                    Parallel.For(0, count,
                        () =>
                        {
                            // GetOrAdd. Use existing open ZipFile or open a new one for this thread.
                            return dictionary.GetOrAdd(Thread.CurrentThread.ManagedThreadId, v =>
                            {
                                return new ZipFile(zipFileName);
                            });
                        },
                        (int i, ParallelLoopState loopState, ZipFile zf) =>
                        {
                            cancellationToken.ThrowIfCancellationRequested();

                            // Get the right entry to extract
                            ZipEntry entry = zf.Entries
                                .Skip(i)
                                .First();

                            // Extract to a file
                            entry.Extract(targetDirectory);

                            return zf;
                        },
                        zf =>
                        {
                        });
                }
                finally
                {
                    // Dispose cached ZipFiles
                    foreach (ZipFile zf in dictionary.Values)
                    {
                        zf.Dispose();
                    }
                }
            } // using
        }

        private static void ExtractAllFilesFromZipFileV5(string zipFileName, string targetDirectory, CancellationToken cancellationToken)
        {
            using (ZipFile zipFile = new ZipFile(zipFileName))
            {
                // Get count of files, files and keep the lock on the file
                ICollection<ZipEntry> zipEntries = zipFile.Entries;
                int count = zipEntries.Where(v => !v.IsDirectory).Count();

                // Caclulate max DOP
                int maxDop = (int)1.5 * Math.Min(count, Environment.ProcessorCount);


                List<Tuple<int, long>> entries = zipEntries
                    .Select((v, i) => Tuple.Create(i, v))
                    .Where(v => !v.Item2.IsDirectory)
                    .Select(v => Tuple.Create(v.Item1, v.Item2.UncompressedSize))
                    .ToList();

                // Load balance between threads
                List<List<Tuple<int, long>>> groupedItems = entries.ToBuckets(maxDop, v => v.Item2 + 10 * 1024 * 1024).ToList();

                // Ensure seq reading from zip file.
                for (int i = 0; i < groupedItems.Count; ++i)
                {
                    groupedItems[i] = groupedItems[i].OrderBy(v => v.Item1).ToList();
                }

                // Cache instances of ZipFile used by threads
                // Make sure that we have open zip file not more than N times, where N is maxDop.
                ConcurrentDictionary<int, Tuple<ZipFile, List<ZipEntry>>> dictionary = new ConcurrentDictionary<int, Tuple<ZipFile, List<ZipEntry>>>(maxDop, maxDop);
                ParallelOptions parallelOptions = new ParallelOptions { MaxDegreeOfParallelism = maxDop, };

                try
                {
                    Parallel.For(0, maxDop, parallelOptions,
                        () =>
                        {
                            // GetOrAdd. Re-use existing open ZipFile or open a new one for this thread.
                            return dictionary.GetOrAdd(Thread.CurrentThread.ManagedThreadId, v =>
                            {
                                ZipFile zf = new ZipFile(zipFileName) { ExtractExistingFile = ExtractExistingFileAction.Throw, FlattenFoldersOnExtract = false, ZipErrorAction = ZipErrorAction.Throw, };
                                zf.ExtractProgress += (sender, e) =>
                                {
                                    cancellationToken.ThrowIfCancellationRequested();
                                };
                                return Tuple.Create(zf, zf.Entries.ToList());
                            });
                        },
                        (int j, ParallelLoopState loopState, Tuple<ZipFile, List<ZipEntry>> zf) =>
                        {

                            List<Tuple<int, long>> list = groupedItems[j];
                            for (int n = 0; n < list.Count; ++n)
                            {
                                cancellationToken.ThrowIfCancellationRequested();

                                int i = list[n].Item1;

                                // Get the right entry to extract
                                ZipEntry entry = zf.Item2[i];
                                Debug.Assert(entry.UncompressedSize == list[n].Item2);

                                // Extract to a file
                                entry.Extract(targetDirectory);
                            }

                            return zf;
                        },
                        zf =>
                        {
                        });
                }
                finally
                {
                    // Dispose cached ZipFiles
                    foreach (Tuple<ZipFile, List<ZipEntry>> zf in dictionary.Values)
                    {
                        try
                        {
                            zf.Item2.Clear();
                            zf.Item1.Dispose();
                        }
                        catch (ZlibException)
                        {
                            // There is a well known defect in Ionic.ZLib
                            // This exception may happen when you read only part of file (not entire file)
                            // and close its handle.
                            // Ionic.Zlib.ZlibException: Bad CRC32 in GZIP trailer. (actual(D202EF8D)!=expected(A39D1010))
                        }
                    }
                }
            }
        }


        private static IEnumerable<List<T>> ToBuckets<T>(this IEnumerable<T> list, int bucketCount, Func<T, long> getWeight)
        {
            List<T> sortedList = list.OrderByDescending(v => getWeight(v)).ToList();

            List<long> runningTotals = Enumerable.Repeat(0L, bucketCount).ToList();
            List<List<T>> buckets = Enumerable.Range(0, bucketCount)
                .Select(v => new List<T>(sortedList.Count / bucketCount))
                .ToList();

            foreach (T item in sortedList)
            {
                // MinBy runningTotal
                int i = runningTotals.IndexOfMin();
                // Add to bucket
                runningTotals[i] += getWeight(item);
                buckets[i].Add(item);
            }

            return buckets;
        }

        public static int IndexOfMin<T>(this IEnumerable<T> source, IComparer<T> comparer = null)
        {
            if (source == null)
            {
                throw new ArgumentNullException(nameof(source));
            }

            if (comparer == null)
            {
                comparer = Comparer<T>.Default;
            }

            using (IEnumerator<T> enumerator = source.GetEnumerator())
            {
                if (!enumerator.MoveNext())
                {
                    return -1; // or maybe throw InvalidOperationException
                }

                int minIndex = 0;
                T minValue = enumerator.Current;

                int index = 0;
                while (enumerator.MoveNext())
                {
                    ++index;
                    if (comparer.Compare(enumerator.Current, minValue) < 0)
                    {
                        minIndex = index;
                        minValue = enumerator.Current;
                    }
                }

                return minIndex;
            }
        }

        public static int IndexOfMinBy<TSource, TKey>(this IEnumerable<TSource> source, Func<TSource, TKey> selector, IComparer<TKey> comparer = null)
        {
            if (source == null)
            {
                throw new ArgumentNullException(nameof(source));
            }

            if (comparer == null)
            {
                comparer = Comparer<TKey>.Default;
            }

            using (IEnumerator<TSource> enumerator = source.GetEnumerator())
            {
                if (!enumerator.MoveNext())
                {
                    return -1; // or maybe throw InvalidOperationException
                }

                int minIndex = 0;
                TKey minValue = selector(enumerator.Current);

                int index = 0;
                while (enumerator.MoveNext())
                {
                    ++index;
                    TKey value = selector(enumerator.Current);
                    if (comparer.Compare(value, minValue) < 0)
                    {
                        minIndex = index;
                        minValue = value;
                    }
                }

                return minIndex;
            }
        }
    }
}

【讨论】:

    【解决方案3】:

    问题是您只用一个句柄打开文件一次。一个句柄有一个读取位置,如果您在同一个句柄上进行并行读取,读取位置会变得混乱。使用多个句柄多次打开文件,应该没问题。

    【讨论】:

      【解决方案4】:

      我同意 Floste 的回答,但我建议采用不同的方法。如果您的文件在 zip 文件中约为 50k:

      1) 为每个文件创建一个字节数组队列。 2) 队列的每个成员都是一个提取的 zip 文件条目。 3) 尝试从 zip 文件中提取文件到字节数组中,并在提取完成后将其添加到队列中。 4)提取线程应该是单线程,没有并行性。 5)当提取线程正在执行它的工作时,创建另一个线程/任务来清空队列。这些任务将从队列中提取数据并将它们写入磁盘。由于它们是不同的文件,因此不会出现竞争条件或资源不可用。

      队列可能需要互斥体或锁。 这可能不是最好的方法,但我相信你会加快速度。

      【讨论】:

        【解决方案5】:

        我需要对一个大型存档(约 30 GB,约 45k 可变大小条目)进行并行解压缩,并利用 DotNetZip 提出了这个解决方案:

            public static void ParallelExtract(
                string archivePath,
                string destinationPath,
                string password,
                CancellationToken token,
                ProgressReportDelegate progress // Could also be Progress<T> or whatever you prefer.
                )
            {
                if (String.IsNullOrEmpty(archivePath))
                    throw new ArgumentNullException("archivePath");
        
                if (String.IsNullOrEmpty(destinationPath))
                    throw new ArgumentNullException("destinationPath");
        
                Stopwatch elapsed = new Stopwatch();
                Stopwatch progressReportingTimer = new Stopwatch();
        
                elapsed.Start();
                progressReportingTimer.Start();
        
                object obj = new object();
        
                int count = -1;
                long bytesExtracted = 0;
                long bytesTotal = -1;
        
                List<Task> decompressors = new List<Task>();
        
                for (int i = 0; i < Environment.ProcessorCount; i++)
                {
                    decompressors.Add(Task.Run(() =>
                    {
                        using (ZipFile zipFile = new ZipFile(archivePath))
                        {
                            if (!String.IsNullOrEmpty(password))
                                zipFile.Password = password;
        
                            zipFile.ExtractProgress += delegate (object zipSender, ExtractProgressEventArgs zipArgs)
                            {
                                // Report progress after each EntryBytesWritten event, as long as it's been at least 250ms since the last report, so as to not overwhelm listeners like a progress bar. 
                                // Fire regardless upon completion (bytesExtracted == bytesTotal) to provide a final update before finishing.
                                if ((zipArgs.EventType == ZipProgressEventType.Extracting_EntryBytesWritten && progressReportingTimer.ElapsedMilliseconds >= 250) || bytesExtracted == bytesTotal)
                                {
                                    int percentage = Percentage(bytesExtracted, bytesTotal);
        
                                    lock (obj)
                                    {
                                        progress?.Invoke(); // <-- Handle your progress updates here.
        
                                        progressReportingTimer.Restart();
                                    }
                                }
                            };
        
                            // Block all threads until we sum the total size of all entries so that when we begin processing on the threadpool we 
                            // can report progress relative to the total.
                            lock (obj)
                                if (bytesTotal == -1)
                                    foreach (var entry in zipFile.Entries)
                                        bytesTotal += entry.CompressedSize;
        
                            var array = zipFile.Entries.ToArray();
        
                            int index;
                            ZipEntry zipEntry;
        
                            // Iterate through the archive's entries sequentially despite being on multiple threads.
                            while (count < zipFile.Entries.Count && !token.IsCancellationRequested)
                            {
                                index = Interlocked.Increment(ref count);
        
                                if (index >= zipFile.Entries.Count)
                                    return;
        
                                zipEntry = array[index];
        
                                Interlocked.Add(ref bytesExtracted, zipEntry.CompressedSize);
        
                                zipEntry.Extract(destinationPath, ExtractExistingFileAction.OverwriteSilently);
                            }
                        }
                    },
                    token
                    ));
                }
        
                Task.WaitAll(decompressors.ToArray());
            }
        

        结果:

        硬件:Intel Core i7-4710HQ @ 3.50 GHz(4 核,8 个超线程),16 GB RAM,(SATA) SSD,Win10x64 1903:

        存档:44.5k 条目,~30 GB DEFLATE(存储)

        Threads:    Time:
        -----------------
        1           35:20 (NOTE: This is virtually identical to ZipArchive.ExtractAll())
        2           22:14
        3           18:40
        4           16:49
        8           14:42
        

        【讨论】:

        • @maytham-ɯɐɥʇʎɐɯ 谢谢;当我编辑 cmets 时,它无意中坏了。
        猜你喜欢
        • 1970-01-01
        • 2012-04-15
        • 1970-01-01
        • 2011-05-02
        • 1970-01-01
        • 2010-12-14
        • 1970-01-01
        • 2017-08-23
        • 1970-01-01
        相关资源
        最近更新 更多