Parallel.ForEach का उपयोग कब करें और कब PLINQ के लिए करें

 public static ParallelLoopResult ForEach<TSource >( IEnumerable<TSource> source, Action<TSource, ParallelLoopState,Int64>body) 

 public static double [] PairwiseMultiply( double[] v1, double[] v2) { var length = Math.Min(v1.Length, v2.Lenth); double[] result = new double[length]; Parallel.ForEach(v1, (element, loopstate, elementIndex) => result[elementIndex] = element * v2[elementIndex]); return result; } 

• पहला विकल्प IEnumerable.Count () को कॉल करना है, जिसकी लागत O (n) होगी। यदि तत्वों की संख्या ज्ञात है, तो आप किसी दिए गए इंडेक्स पर परिणामों को बचाने के लिए एक आउटपुट सरणी बना सकते हैं
• दूसरा विकल्प संग्रह को भौतिक बनाना है (इसे मोड़कर, उदाहरण के लिए, एक सरणी में)। यदि बहुत अधिक डेटा है, तो यह विधि बहुत उपयुक्त नहीं है।
• तीसरा विकल्प आउटपुट संग्रह के बारे में सावधानी से सोचना है। आउटपुट संग्रह एक हैश हो सकता है, फिर आउटपुट मान को संग्रहीत करने के लिए आवश्यक मेमोरी की मात्रा हैश टकराव से बचने के लिए इनपुट मेमोरी से कम से कम 2 गुना होगी; यदि बहुत अधिक डेटा है, तो हैश के लिए डेटा संरचना निषेधात्मक रूप से बड़ी होगी, इसके अलावा, आप गलत साझाकरण और कचरा संग्रहकर्ता के कारण एक प्रदर्शन ड्रॉप प्राप्त कर सकते हैं।
• और अंतिम विकल्प उनके मूल अनुक्रमित के साथ परिणामों को सहेजना है, और फिर आउटपुट संग्रह को सॉर्ट करने के लिए अपना स्वयं का एल्गोरिदम लागू करना है।

समानांतर स्ट्रीमिंग

एक स्ट्रीम को संसाधित करने के लिए PLINQ का उपयोग करना

• 1. परिणाम सरणी में नहीं चलते हैं, इसलिए मेमोरी में डेटा संग्रहीत करने में कोई अतिरेक नहीं है।
• 2. जब आप नया डेटा प्राप्त करते हैं तो आप गणना की एक ही धारा में परिणाम की गणना कर सकते हैं।

 public static void AnalyzeStocks(IEnumerable<Stock> Stocks) { var StockRiskPortfolio = Stocks .AsParallel() .AsOrdered() .Select(stock => new { Stock = stock, Risk = ComputeRisk(stock)}) .Where(stockRisk => ExpensiveRiskAnalysis(stockRisk.Risk)); foreach (var stockRisk in StockRiskPortfolio) { SomeStockComputation(stockRisk.Risk); // StockRiskPortfolio will be a stream of results } } 

• ParallelMergeOptions.NotBuffered - इंगित करता है कि प्रत्येक संसाधित आइटम को संसाधित होते ही प्रत्येक थ्रेड से वापस कर दिया जाता है।
• ParallelMergeOptions.AutoBuffered - इंगित करता है कि तत्वों को बफर में एकत्र किया जाता है, बफर समय-समय पर उपभोक्ता स्ट्रीम में वापस आ जाता है
• ParallelMergeOptions.FullyBuffered - इंगित करता है कि आउटपुट अनुक्रम पूरी तरह से बफ़र है, यह आपको अन्य विकल्पों का उपयोग करने की तुलना में तेज़ी से परिणाम प्राप्त करने की अनुमति देता है, लेकिन फिर उपभोक्ता थ्रेड को प्रसंस्करण के लिए पहला तत्व प्राप्त करने के लिए लंबा इंतजार करना होगा।

समानांतर क्यों नहीं।फोर ईच?

 public static void AnalyzeStocks(IEnumerable<Stock> Stocks) { Parallel.ForEach(Stocks, stock => { var risk = ComputeRisk(stock); if(ExpensiveRiskAnalysis(risk) { // stream processing lock(myLock) { SomeStockComputation(risk) }; // store results } } 

दो संग्रह पर संचालन

दो संग्रहों पर संचालन के लिए PLINQ का उपयोग करना

 public static IEnumerable<T> Zipping<T>(IEnumerable<T> a, IEnumerable<T> b) { return a .AsParallel() .AsOrdered() .Select(element => ExpensiveComputation(element)) .Zip( b .AsParallel() .AsOrdered() .Select(element => DifferentExpensiveComputation(element)), (a_element, b_element) => Combine(a_element,b_element)); } 

समानांतर क्यों नहीं।फोर ईच?

 public static IEnumerable<T> Zipping<T>(IEnumerable<T> a, IEnumerable<T> b) { var numElements = Math.Min(a.Count(), b.Count()); var result = new T[numElements]; Parallel.ForEach(a, (element, loopstate, index) => { var a_element = ExpensiveComputation(element); var b_element = DifferentExpensiveComputation(b.ElementAt(index)); result[index] = Combine(a_element, b_element); }); return result; } 

धागा-स्थानीय राज्य

ParallelForEach का उपयोग करके स्ट्रीम की स्थानीय स्थिति तक पहुँच प्राप्त करें

 public static ParallelLoopResult ForEach<TSource,TLocal>( IEnumerable<TSource> source, Func<TLocal> localInit, Func<TSource, ParallelLoopState, TLocal, TLocal> body, Action<TLocal> localFinally) 

 public static List<R> Filtering<T,R>(IEnumerable<T> source) { var results = new List<R>(); using (SemaphoreSlim sem = new SemaphoreSlim(1)) { Parallel.ForEach(source, () => new List<R>(), (element, loopstate, localStorage) => { bool filter = filterFunction(element); if (filter) localStorage.Add(element); return localStorage; }, (finalStorage) => { lock(myLock) { results.AddRange(finalStorage) }; }); } return results; } 

 public static void UnsafeDownloadUrls () { WebClient webclient = new WebClient(); Parallel.ForEach(urls, (url,loopstate,index) => { webclient.DownloadFile(url, filenames[index] + ".dat"); Console.WriteLine("{0}:{1}", Thread.CurrentThread.ManagedThreadId, url); }); } 

 public static void BAD_DownloadUrls () { Parallel.ForEach(urls, (url,loopstate,index) => { WebClient webclient = new WebClient(); webclient.DownloadFile(url, filenames[index] + ".dat"); Console.WriteLine("{0}:{1}", Thread.CurrentThread.ManagedThreadId, url); }); } 

 public static void downloadUrlsSafe() { Parallel.ForEach(urls, () => new WebClient(), (url, loopstate, index, webclient) => { webclient.DownloadFile(url, filenames[index]+".dat"); Console.WriteLine("{0}:{1}", Thread.CurrentThread.ManagedThreadId, url); return webclient; }, (webclient) => { }); } } 

यहाँ PLINQ खराब क्यों है?

 public static void downloadUrl() { var webclient = new ThreadLocal<WebClient>(()=> new WebClient ()); var res = urls .AsParallel() .ForAll( url => { webclient.Value.DownloadFile(url, host[url] +".dat")); Console.WriteLine("{0}:{1}", Thread.CurrentThread.ManagedThreadId, url); }); } 

ऑपरेशन से बाहर निकलें

परिचालनों के बाहर निकलने के लिए Parallel.ForEach का उपयोग करना

 public static ParallelLoopResult ForEach<TSource >( IEnumerable<TSource> source, Action<TSource, ParallelLoopState> body) 

ParallelLoopState.Stop ()

 public static boolean FindAny<T,T>(IEnumerable<T> TSpace, T match) where T: IEqualityComparer<T> { var matchFound = false; Parallel.ForEach(TSpace, (curValue, loopstate) => { if (curValue.Equals(match) ) { matchFound = true; loopstate.Stop(); } }); return matchFound; } 

ParallelLoopState.Break ()

 public static int FindLowestIndex<T,T>(IEnumerable<T> TSpace, T match) where T: IEqualityComparer<T> { var loopResult = Parallel.ForEach(source, (curValue, loopState, curIndex) => { if (curValue.Equals(match)) { loopState.Break(); } }); var matchedIndex = loopResult.LowestBreakIteration; return matchedIndex.HasValue ? matchedIndex : -1; } 

PLINQ क्यों नहीं?

निष्कर्ष

उपयोगी लिंक: