इनपुट स्ट्रीम पर अवरुद्ध कॉल को समाप्त नहीं करने में थ्रेड बाधित

Question

मैं एक सीरियल पोर्ट से डेटा पढ़ने के लिए RXTX का उपयोग कर रहा हूं। पढ़ने एक धागा निम्नलिखित तरीके से पैदा की भीतर किया जाता है:

CommPortIdentifier portIdentifier = CommPortIdentifier.getPortIdentifier(port); 
CommPort comm = portIdentifier.open("Whatever", 2000); 
SerialPort serial = (SerialPort)comm; 
...settings 
Thread t = new Thread(new SerialReader(serial.getInputStream())); 
t.start(); 

SerialReader वर्ग Runnable लागू करता है और सिर्फ अनिश्चित काल के लिए लूप होता है, बंदरगाह से पढ़ने और अन्य अनुप्रयोगों के लिए इसे बंद कर भेजने से पहले उपयोगी संकुल में डेटा का निर्माण। हालांकि, मैं यह निम्नलिखित सादगी के लिए नीचे कम किया है:

public void run() { 
    ReadableByteChannel byteChan = Channels.newChannel(in); //in = InputStream passed to SerialReader 
    ByteBuffer buffer = ByteBuffer.allocate(100); 
    while (true) { 
    try { 
     byteChan.read(buffer); 
    } catch (Exception e) { 
     System.out.println(e); 
    } 
    } 
} 

एक उपयोगकर्ता एक स्टॉप बटन, निम्नलिखित कार्यक्षमता आग है कि सिद्धांत रूप में इनपुट धारा को बंद करने और अवरुद्ध byteChan.read से बाहर तोड़ चाहिए (क्लिक करता है बफर) कॉल करें।

public void stop() { 
    t.interrupt(); 
    serial.close(); 
} 

हालांकि, जब मैं इस कोड चलाने के लिए, मैं कभी नहीं एक ClosedByInterruptException, जो एक बार इनपुट स्ट्रीम बंद कर देता है सक्रिय किया जाना चाहिए मिलती है: कोड इस प्रकार है। इसके अलावा, serial.close() पर कॉल पर निष्पादन ब्लॉक - क्योंकि अंतर्निहित इनपुट स्ट्रीम अभी भी पढ़ने कॉल पर अवरुद्ध है। मैंने बाइटकहान.क्लोज़() के साथ इंटरप्ट कॉल को बदलने की कोशिश की है, जिसके बाद एक असिंक्रोनस क्लोज़ एक्सेप्शन का कारण बनना चाहिए, हालांकि, मुझे वही परिणाम मिल रहे हैं।

मुझे जो भी याद आ रही है उसमें कोई मदद की सराहना की जाएगी।

Answer 1

RXTX SerialInputStream (क्या serial.getInputStream() कॉल द्वारा दिया जाता है) एक समय समाप्ति योजना है कि समाप्त हो गया का समर्थन करता है मेरी सभी समस्याओं को हल करना। बनाने नई SerialReader वस्तु नहीं रह अनिश्चित काल के लिए ब्लॉक करने के लिए पढ़ता है का कारण बनता है पहले निम्नलिखित जोड़ना:

serial.enableReceiveTimeout(1000); 

SerialReader वस्तु के भीतर, मैं कुछ चीजें आसपास बदलने के लिए बजाय ReadableByteChannel बनाने की InputStream से सीधे पढ़ा था, लेकिन अब, मैं बिना किसी समस्या के पाठक को रोक और पुनरारंभ कर सकता हूं।

Answer 2

आप एक धारा है कि एक InterruptibleChannel में व्यवधान कारक आई/ओ का समर्थन नहीं करता बस इसे लपेटकर नहीं कर सकते हैं (वैसे भी और,, ReadableByteChannelInterruptibleChannel विस्तार नहीं करता है)।

आपको अंतर्निहित InputStream के अनुबंध को देखना होगा। SerialPort.getInputStream() इसके परिणाम की बाधा के बारे में क्या कहता है? यदि यह कुछ भी नहीं कहता है, तो आपको यह मानना ​​चाहिए कि यह इंटरप्ट को अनदेखा करता है।

किसी भी I/O के लिए जो स्पष्ट रूप से अंतःक्रियाशीलता का समर्थन नहीं करता है, केवल एकमात्र विकल्प आम तौर पर किसी अन्य थ्रेड से स्ट्रीम को बंद कर रहा है। यह धारा में कॉल पर अवरुद्ध धागे में IOException (हालांकि यह AsynchronousCloseException नहीं हो सकता है) को तुरंत बढ़ा सकता है।

हालांकि, यह InputStream — के कार्यान्वयन पर भी निर्भर है और अंतर्निहित ओएस भी एक कारक हो सकता है।

---

नोट ReadableByteChannelImpl वर्ग newChannel() द्वारा दिया पर स्रोत कोड टिप्पणी:

private static class ReadableByteChannelImpl 
    extends AbstractInterruptibleChannel  // Not really interruptible 
    implements ReadableByteChannel 
    { 
    InputStream in; 
    ⋮ 

Answer 3

मैं rxtx को बंद करने के लिए नीचे दिए गए कोड का उपयोग कर रहा हूं। मैं उन परीक्षणों को चलाता हूं जो उन्हें शुरू करते हैं और उन्हें बंद करते हैं और ऐसा लगता है कि ठीक काम करता है।मेरे पाठक लगता है:

private void addPartsToQueue(final InputStream inputStream) { 
    byte[] buffer = new byte[1024]; 
    int len = -1; 
    boolean first = true; 
    // the read can throw 
    try { 
     while ((len = inputStream.read(buffer)) > -1) { 
      if (len > 0) { 
       if (first) { 
        first = false; 
        t0 = System.currentTimeMillis(); 
       } else 
        t1 = System.currentTimeMillis(); 
       final String part = new String(new String(buffer, 0, len)); 
       queue.add(part); 
       //System.out.println(part + " " + (t1 - t0)); 
      } 
      try { 
       Thread.sleep(sleep); 
      } catch (InterruptedException e) { 
       //System.out.println(Thread.currentThread().getName() + " interrupted " + e); 
       break; 
      } 
     } 
    } catch (IOException e) { 
     System.err.println(Thread.currentThread().getName() + " " + e); 
     //if(interruSystem.err.println(e); 
     e.printStackTrace(); 
    } 
    //System.out.println(Thread.currentThread().getName() + " is ending."); 
} 

धन्यवाद

public void shutdown(final Device device) { 
    shutdown(serialReaderThread); 
    shutdown(messageAssemblerThread); 
    serialPort.close(); 
    if (device != null) 
     device.setSerialPort(null); 
} 

public static void shutdown(final Thread thread) { 
    if (thread != null) { 
     //System.out.println("before intterupt() on thread " + thread.getName() + ", it's state is " + thread.getState()); 
     thread.interrupt(); 
     //System.out.println("after intterupt() on thread " + thread.getName() + ", it's state is " + thread.getState()); 
     try { 
      Thread.sleep(100); 
     } catch (InterruptedException e) { 
      System.out.println(Thread.currentThread().getName() + " was interrupted trying to sleep after interrupting" + thread.getName() + " " + e); 
     } 
     //System.out.println("before join() on thread " + thread.getName() + ", it's state is " + thread.getState()); 
     try { 
      thread.join(); 
     } catch (InterruptedException e) { 
      System.out.println(Thread.currentThread().getName() + " join interruped"); 
     } 
     //System.out.println(Thread.currentThread().getName() + " after join() on thread " + thread.getName() + ", it's state is" + thread.getState()); 
    }