C# अंतर-धागा संचार

Question

हैलो मैं एक निर्माता और एक उपभोक्ता सहयोग करने के लिए धागे करना चाहते हैं। उपभोक्ता धीमा है, और निर्माता बहुत तेज है और विस्फोटों में काम करता है।

उदाहरण के लिए उपभोक्ता 20 सेकंड प्रति एक संदेश संसाधित कर सकते हैं, और निर्माता एक सेकंड में 10 संदेश उत्पादन कर सकते हैं, लेकिन यह एक लंबे समय के बारे में एक बार होता है, जबकि इतने उपभोक्ता मिल सकते हैं।

मैं चाहता हूँ की तरह कुछ:

Stream commonStream; 
AutoResetEvent commonLock; 

void Producer() 
{ 
    while (true) 
    { 
    magic.BlockUntilMagicAvalible(); 
    byte[] buffer = magic.Produce(); 
    commonStream.Write(buffer); 
    commonLock.Set(); 
    } 
} 

void Consumer() 
{ 
    while(true) 
    { 
    commonLock.WaitOne(); 
    MagicalObject o = binarySerializer.Deserialize(commonStream); 
    DoSomething(o); 
    } 
} 

Answer 1

मैं निम्नलिखित लेख पढ़ने होगा कि वे अपनी समस्या का वर्णन। असल में आपको काम की अपनी इकाई के लिए सही अलगाव नहीं मिल रहा है।

http://blogs.msdn.com/b/ricom/archive/2006/04/24/582643.aspx http://blogs.msdn.com/b/ricom/archive/2006/04/26/584802.aspx

Answer 2

आप नेट 4.0 या उच्चतर है, तो आप इसे एक BlockingCollection

int maxBufferCap = 500; 
BlockingCollection<MagicalObject> Collection 
          = new BlockingCollection<MagicalObject>(maxBufferCap); 
void Producer() 
{ 
    while (magic.HasMoreMagic) 
    { 
     this.Collection.Add(magic.ProduceMagic()); 
    } 
    this.Collection.CompleteAdding(); 
} 

void Consumer() 
{ 
    foreach (MagicalObject magicalObject in this.Collection.GetConsumingEnumerable()) 
    { 
     DoSomthing(magicalObject); 
    } 
} 

foreach लाइन का उपयोग करता है, तो वहाँ बफर में कोई डाटा नहीं है सो जाएगा द्वारा इस तरह से कर सकते हैं, जब संग्रह में कुछ जोड़ा जाता है तो यह स्वचालित रूप से इसे स्वयं भी जगाएगा।

कारण मैं अधिकतम बफर सेट करते हैं तो अपने निर्माता उपभोक्ता आप स्मृति का एक बहुत लेने के रूप में अधिक से अधिक वस्तुओं संग्रह करने के लिए में डाल करने के अंत हो सकता है की तुलना में बहुत तेजी से होता है। जैसा कि आप अवरुद्ध संग्रह जब बफर आकार तक एक आइटम उपभोक्ता द्वारा संग्रह से हटा दिया गया अवरुद्ध कर देगा निर्माता पर Add कॉल तक पहुँच जाता है बनाने के एक अधिकतम बफ़र आकार सेट करके।

BlockingCollection कक्षा का एक और बोनस यह है कि इसमें जितने चाहें उतने उत्पादक और उपभोक्ता हो सकते हैं, इसे 1: 1 अनुपात होने की आवश्यकता नहीं है। DoSomthing इसका समर्थन करता है, तो आप (या यहां तक ​​Parallel.ForEach का उपयोग करें और डेटा स्रोत के रूप गणनीय लेने का उपयोग)

void ConsumersInParalell() 
{ 
    //This assumes the method signature of DoSomthing is one of the following: 
    // Action<MagicalObject> 
    // Action<MagicalObject, ParallelLoopState> 
    // Action<MagicalObject, ParallelLoopState, long> 
    Paralell.ForEach(this.Collection.GetConsumingEnumerable(), DoSomthing); 
} 

Answer 3

आप क्या आप एक कतार और टाइमर का उपयोग करना चाहते हैं प्राप्त कर सकते हैं कंप्यूटर के मूल प्रति एक foreach पाश हो सकता था। निर्माता कतार में मूल्य जोड़ता है और उपभोक्ता टाइमर शुरू करता है। उपभोक्ता टाइमर का बीता घटना (जो एक ThreadPool धागे पर है) टाइमर बंद हो जाता है, और कतार के माध्यम से लूप जब तक यह खाली है, तो गायब हो जाता है (कोई अनावश्यक मतदान)। उपभोक्ता अभी भी चल रहा है, जबकि निर्माता कतार में जोड़ सकते हैं।

System.Timers.Timer consumerTimer; 
Queue<byte[]> queue = new Queue<byte[]>(); 

void Producer() 
{ 
    consumerTimer = new System.Timers.Timer(1000); 
    consumerTimer.Elapsed += new System.Timers.ElapsedEventHandler(consumerTimer_Elapsed); 
    while (true) 
    { 
     magic.BlockUntilMagicAvailable(); 
     lock (queue) 
     { 
      queue.Enqueue(magic.Produce()); 
      if (!consumerTimer.Enabled) 
      { 
       consumerTimer.Start(); 
      } 
     } 
    } 
} 

void consumerTimer_Elapsed(object sender, System.Timers.ElapsedEventArgs e) 
{ 
    while (true) 
    { 
     consumerTimer.Stop(); 
     lock (queue) 
     { 
      if (queue.Count > 0) 
      { 
       DoSomething(queue.Dequeue()); 
      } 
      else 
      { 
       break; 
      } 
     } 
    } 
} 

Answer 4

मैं म्यूटेक्स का उपयोग करता हूं। विचार यह है कि दोनों अलग-अलग धागे में चलते हैं। कंज्यूमर थ्रेड एक म्यूटेक्स द्वारा लॉक किया गया है, जहां यह निर्माता द्वारा रिलीज होने तक अनिश्चित काल तक बैठेगा। इसके बाद निर्माता को जारी रखने के लिए समानांतर में डेटा को संसाधित किया जाएगा। पूर्ण होने पर उपभोक्ता फिर से लॉक हो जाएगा।

(कोड धागा शुरू करते हैं, और अन्य गुणवत्ता बिट्स संक्षिप्तता के लिए छोड़ दिए गए हैं।)

// Pre-create mutex owned by Producer thread, then start Consumer thread. 
Mutex mutex = new Mutex(true); 
Queue<T> queue = new Queue<T>(); 

void Producer_AddData(T data) 
{ 
    lock (queue) { 
    queue.Enqueue(GetData()); 
    } 

    // Release mutex to start thread: 
    mutex.ReleaseMutex(); 
    mutex.WaitOne(); 
} 

void Consumer() 
{ 
    while(true) 
    { 
    // Wait indefinitely on mutex 
    mutex.WaitOne(); 
    mutex.ReleaseMutex(); 

    T data; 
    lock (queue) { 
     data = queue.Dequeue(); 
    } 
    DoSomething(data); 
    } 

}

यह एक बहुत ही कुछ मिलीसेकेंड उपभोक्ता जगाने के लिए प्रतीक्षा करता है है, जबकि द्वारा निर्माता धीमा कर देती है और म्यूटेक्स को छोड़ दें। यदि आप इसके साथ रह सकते हैं।