एक छोटे से उदाहरण के साथ, मैं आपको 15 मिनट में मल्टी-थ्रेडेड फ़ाइल संपीड़न के गैर-तुच्छ कार्य को हल करने के लिए टीपीएल डेटाफ्लो लाइब्रेरी का उपयोग करने का तरीका बताऊंगा।

कार्य

System.IO.Compression नाम स्थान में स्थित GZipStream वर्ग का उपयोग करके प्रभावी फ़ाइल संपीड़न को लागू करना आवश्यक है। यह माना जाता है कि हम बड़ी फ़ाइलों को संपीड़ित करेंगे, और वे पूरी तरह से रैम में फिट नहीं हो सकते हैं।

Tpl डेटाफ़्लो

TPL Dataflow (TDF) को .NET 4 में शामिल TPL (टास्क पैरेलल लाइब्रेरी) लाइब्रेरी के शीर्ष पर बनाया गया है और स्रोत लाइब्रेरी की तुलना में अधिक जटिल समस्याओं को हल करने के लिए प्राइमेटिक्स के एक सेट के साथ इसे पूरक करता है। TPL Dataflow डेटा स्ट्रीम के समानांतर प्रसंस्करण के लिए समर्थन जोड़ने के लिए कार्यों, थ्रेड-सुरक्षित संग्रह और .NET 4 में शुरू की गई अन्य विशेषताओं का उपयोग करता है। लाइब्रेरी का सार विभिन्न डेटा प्रोसेसिंग चेन को व्यवस्थित करने के लिए विभिन्न ब्लॉकों को जोड़ना है। इसके अलावा, डेटा प्रोसेसिंग सिंक्रोनस और एसिंक्रोनस दोनों तरह से हो सकती है। लाइब्रेरी को आगामी .NET 4.5 में शामिल किया जाएगा।

निर्णय

इस समस्या को हल करने के लिए, आपको केवल 3 ब्लॉक चाहिए:

1. डेटा स्रोत से पढ़े गए डेटा के लिए बफर:
   

    var buffer = new BufferBlock<byte[]>(); 

   
   
2. डेटा संपीड़न इकाई:
   

    var compressor = new TransformBlock<byte[], byte[]>(bytes => Compress(bytes)); 

   
   संपीड़न समारोह:
   

    private static byte[] Compress(byte[] bytes) { using (var resultStream = new MemoryStream()) { using (var zipStream = new GZipStream(resultStream, CompressionMode.Compress)) { using (var writer = new BinaryWriter(zipStream)) { writer.Write(bytes); return resultStream.ToArray(); } } } } 

   
   
3. संकुचित डेटा रिकॉर्डिंग इकाई:
   

    var writer = new ActionBlock<byte[]>(bytes => outputStream.Write(bytes, 0, bytes.Length)); 

   
   

हमारे ब्लॉक कनेक्ट करें:

 buffer.LinkTo(compressor); compressor.LinkTo(writer); 

हम अपने ब्लॉकों को भी सूचित करेंगे जब उनके लिए डेटा समाप्त हो गया है और वे अपना काम पूरा कर सकते हैं। आप ब्लॉक की Complete विधि को कॉल करके ऐसा कर सकते हैं:

 buffer.Completion.ContinueWith(task => compressor.Complete()); compressor.Completion.ContinueWith(task => writer.Complete()); 

जैसे ही हम फ़ाइल पढ़ते हैं, हम अपने बफर को डेटा प्रदान करेंगे। यह ब्लॉक की Post विधि को कॉल करके किया जाता है:

 while (!buffer.Post(bytes)) { } 

ब्लॉक पूरा होने पर स्थिति को ध्यान में रखने के लिए हमें इस तरह के निर्माण की आवश्यकता है और अब डेटा प्राप्त नहीं होता है।

पढ़ने के पूरा होने पर, हम अपने ब्लॉक को सूचित करेंगे कि हमने डेटा बाहर चला दिया है:

 buffer.Complete(); 

अब हमें बस अपने ब्लॉक के अंत तक इंतजार करना होगा, जो स्ट्रीम में कंप्रेस्ड डेटा लिखने के लिए जिम्मेदार है:

 writer.Completion.Wait(); 

परिणामी विधि:

 public static void Compress(Stream inputStream, Stream outputStream) { var buffer = new BufferBlock<byte[]>(); var compressor = new TransformBlock<byte[], byte[]>(bytes => Compress(bytes)); var writer = new ActionBlock<byte[]>(bytes => outputStream.Write(bytes, 0, bytes.Length)); buffer.LinkTo(compressor); buffer.Completion.ContinueWith(task => compressor.Complete()); compressor.LinkTo(writer); compressor.Completion.ContinueWith(task => writer.Complete()); var readBuffer = new byte[BufferSize]; while (true) { int readCount = inputStream.Read(readBuffer, 0, BufferSize); if (readCount > 0) { var bytes = new byte[readCount]; Buffer.BlockCopy(readBuffer, 0, bytes, 0, readCount); while (!buffer.Post(bytes)) { } } if (readCount != BufferSize) { buffer.Complete(); break; } } writer.Completion.Wait(); } 

हम इसे समाप्त कर सकते हैं, यदि एक "लेकिन" के लिए नहीं: यह कोड बिल्कुल तुल्यकालिक एक से गति में भिन्न नहीं है। इसके लिए तेजी से काम करने के लिए, हमें यह इंगित करने की आवश्यकता है कि हमारे संपीड़न ऑपरेशन को एसिंक्रोनस रूप से किया जाना चाहिए। आप हमारे ब्लॉक में आवश्यक सेटिंग्स जोड़कर ऐसा कर सकते हैं:

 var compressorOptions = new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 4 }; var compressor = new TransformBlock<byte[], byte[]>(bytes => Compress(bytes), compressorOptions); 

हमें ऐसी स्थिति को भी समझने की आवश्यकता है जहां डेटा संपीड़ित की तुलना में तेजी से पढ़ा जाता है या संकुचित की तुलना में धीमा लिखा जाता है। आप यह कर सकते हैं हमारे ब्लॉक की BoundedCapacity संपत्ति को बदलकर:

 var buffer = new BufferBlock<byte[]>(new DataflowBlockOptions { BoundedCapacity = 100 }); var compressorOptions = new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 4, BoundedCapacity = 100 }; var compressor = new TransformBlock<byte[], byte[]>(bytes => Compress(bytes), compressorOptions); var writerOptions = new ExecutionDataflowBlockOptions { BoundedCapacity = 100, SingleProducerConstrained = true }; var writer = new ActionBlock<byte[]>(bytes => outputStream.Write(bytes, 0, bytes.Length), writerOptions); 

अंतिम विधि इस प्रकार है:

 public static void Compress(Stream inputStream, Stream outputStream) { var buffer = new BufferBlock<byte[]>(new DataflowBlockOptions {BoundedCapacity = 100}); var compressorOptions = new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 4, BoundedCapacity = 100 }; var compressor = new TransformBlock<byte[], byte[]>(bytes => Compress(bytes), compressorOptions); var writerOptions = new ExecutionDataflowBlockOptions { BoundedCapacity = 100, SingleProducerConstrained = true }; var writer = new ActionBlock<byte[]>(bytes => outputStream.Write(bytes, 0, bytes.Length), writerOptions); buffer.LinkTo(compressor); buffer.Completion.ContinueWith(task => compressor.Complete()); compressor.LinkTo(writer); compressor.Completion.ContinueWith(task => writer.Complete()); var readBuffer = new byte[BufferSize]; while (true) { int readCount = inputStream.Read(readBuffer, 0, BufferSize); if (readCount > 0) { var postData = new byte[readCount]; Buffer.BlockCopy(readBuffer, 0, postData, 0, readCount); while (!buffer.Post(postData)) { } } if (readCount != BufferSize) { buffer.Complete(); break; } } writer.Completion.Wait(); } 

हम इसे ऐसे कंसोल एप्लिकेशन से उदाहरण के लिए कह सकते हैं:

 private const int BufferSize = 16384; static void Main(string[] args) { var stopwatch = Stopwatch.StartNew(); using (var inputStream = File.OpenRead(@"C:\file.bak")) { using (var outputStream = File.Create(@"E:\file.gz")) { Compress(inputStream, outputStream); } } stopwatch.Stop(); Console.WriteLine(); Console.WriteLine(string.Format("Time elapsed: {0}s", stopwatch.Elapsed.TotalSeconds)); Console.ReadKey(); } 

निष्कर्ष

जैसा कि आप देख सकते हैं, टीपीएल डेटाफ्लो का उपयोग बहु-थ्रेडेड डेटा प्रोसेसिंग कार्यों को बहुत सरल कर सकता है। मेरे परीक्षणों में, संपीड़न के लिए आवश्यक समय लगभग 3 गुना कम हो गया था।
आप इस पुस्तकालय को डाउनलोड कर सकते हैं और आधिकारिक पृष्ठ पर इसके बारे में पढ़ सकते हैं।