निर्माता-उपभोक्ता इंटर-थ्रेड संचार

Question

इंटर-थ्रेड संचार में परेशानी हो रही है और पूरे स्थान पर "डमी संदेश" का उपयोग कर इसे हल किया गया है। क्या यह एक बुरा विचार है? संभावित समाधान क्या हैं?

उदाहरण समस्या मेरे पास है।

मुख्य धागा डेटाबेस में रिकॉर्ड को प्रोसेस करने और डालने के लिए धागा शुरू करता है। मुख्य धागा एक संभावित फ़ाइल को पढ़ता है और एक ब्लॉकिंगक्यू में एक के बाद एक रिकॉर्ड (ऑब्जेक्ट) रखता है। प्रोसेसिंग थ्रेड कतार से पढ़ता है और काम करता है।

मैं "प्रसंस्करण थ्रेड" को रोकने के लिए कैसे कहूं? कतार खाली हो सकती है लेकिन काम नहीं किया जाता है और मुख्य धागा या तो थ्रेड प्रसंस्करण के काम को समाप्त नहीं करता है और इसे बाधित नहीं कर सकता है।

तो प्रसंस्करण धागा

while (queue.size() > 0 || !Thread.currentThread().isInterrupted()) { 
    MyObject object= queue.poll(100, TimeUnit.MILLISECONDS); 
    if (object != null) { 
     String data = object.getData(); 
     if (data.equals("END")) { 
      break; 
     } 
    // do work 
    } 
} 
// clean-up 
synchronized queue) { 
    queue.notifyAll(); 
} 
return; 

और मुख्य थ्रेड करता

// ...start processing thread... 
while(reader.hasNext(){ 
    // ...read whole file and put data in queue... 
} 
MyObject dummy = new MyObject(); 
dummy.setData("END"); 
queue.put(dummy); 
//Note: empty queue here means work is done 
while (queue.size() > 0) { 
    synchronized (queue) { 
     queue.wait(500); // over-cautios locking prevention i guess 
    } 
} 

ध्यान दें कि प्रविष्टि ही लेन-देन में होना चाहिए और लेन-देन मुख्य थ्रेड से संभाला नहीं जा सकता।

ऐसा करने का बेहतर तरीका क्या होगा? (मैं सीख रहा हूँ और शुरू करने के लिए "इसे गलत तरीके से कर रही है" नहीं करना चाहती)

Answer 1

ये डमी संदेश मान्य है। इसे "जहर" कहा जाता है। ऐसा कुछ जो निर्माता इसे रोकने के लिए उपभोक्ता को भेजता है।

अन्य संभावना) कहीं मुख्य थ्रेड में Thread.interrupt (कॉल करने के लिए और पकड़ और उसके अनुसार InterruptedException संभाल, कार्यकर्ता सूत्र में है।

Answer 2

जगह पर "डमी संदेश" का उपयोग करके इसे हल किया गया। यह एक बुरा विचार है? संभावित समाधान क्या हैं?

यह बुरा विचार नहीं है, इसे "जहर पिल्स" कहा जाता है और यह थ्रेड-आधारित सेवा को रोकने का एक उचित तरीका है।

लेकिन यह केवल जब उत्पादकों और उपभोक्ताओं की संख्या में जाना जाता है काम करता है।

कोड आप पोस्ट में, दो धागे देखते हैं, एक "मुख्य थ्रेड" है, जो डेटा का उत्पादन होता है, अन्य "प्रसंस्करण धागा" है, जो डेटा की खपत है, "जहर की गोलियों" इस परिस्थिति के लिए अच्छी तरह से काम करता है।

लेकिन कल्पना करने के लिए, यदि आपके पास अन्य उत्पादक भी हैं, तो उपभोक्ता को कैसे रोकना है (केवल तभी जब सभी उत्पादक "जहर की गोलियाँ" भेजते हैं), आपको सभी उत्पादकों की संख्या जानने और जांचने की आवश्यकता है "जहर की गोलियों" उपभोक्ता की संख्या में, अगर यह उत्पादकों की संख्या, जिसका अर्थ है सभी उत्पादकों काम करना बंद कर के बराबर है, तो उपभोक्ता बंद हो जाता है।

"मुख्य धागे" में, आपको InterruptedException पकड़ने की आवश्यकता है, यदि नहीं, तो "मुख्य धागा" "जहर पिल्ल" सेट करने में सक्षम नहीं हो सकता है। आप

... 
try { 
    // do normal processing 
} catch (InterruptedException e) { /* fall through */ } 
finally { 
    MyObject dummy = new MyObject(); 
    dummy.setData("END"); 
    ... 
} 
... 

भी नीचे की तरह यह कर सकते हैं, आप अपने सभी समस्या को हल करने ExecutorService इस्तेमाल करने की कोशिश कर सकते हैं।

(यह काम करता है जब आप बस कुछ काम करता है और फिर जब सभी कार्य पूरा हो जाएगा रोकने की आवश्यकता)

void doWorks(Set<String> works, long timeout, TimeUnit unit) 
    throws InterruptedException { 
    ExecutorService exec = Executors.newCachedThreadPool(); 
    try { 
     for (final String work : works) 
      exec.execute(new Runnable() { 
        public void run() { 
         ... 
        } 
       }); 
    } finally { 
     exec.shutdown(); 
     exec.awaitTermination(timeout, unit); 
    } 
} 

मैं सीख रहा हूँ और "इसे गलत तरीके से कर रही है" शुरू करने के लिए नहीं करना चाहते हैं

आपको पुस्तक को पढ़ने की आवश्यकता हो सकती है: प्रैक्टिस में जावा कंसुरेंसी। मेरा विश्वास करो, यह सबसे अच्छा है।

Answer 3

आप क्या कर सकते हैं (जो मैंने हाल ही में प्रोजेक्ट में किया था) कतार को लपेटना है और फिर 'isOpen()' विधि जोड़ें।

class ClosableQ<T> { 

    boolean isOpen = true; 

    private LinkedBlockingQueue<T> lbq = new LinkedBlockingQueue<T>(); 

    public void put(T someObject) { 
     if (isOpen) { 
     lbq.put(someObject); 
     } 
    } 

    public T get() { 
     if (isOpen) { 
     return lbq.get(0); 
     } 
    } 

    public boolean isOpen() { 
     return isOpen; 
    } 

    public void open() { 
     isOpen = true; 
    } 

    public void close() { 
     isOpen = false; 
    } 
} 

तो अपने लेखक धागा हो जाता है की तरह कुछ:

while (reader.hasNext()) { 
    // read the file and put it into the queue 
    dataQ.put(someObject); 
} 
// now we're done 
dataQ.close(); 

और पाठक धागा:

while (dataQ.isOpen) { 
    someObject = dataQ.get(); 
} 

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