Ngắt chuỗi không kết thúc cuộc gọi chặn trên luồng đầu vào đọc

Question

Tôi đang sử dụng RXTX để đọc dữ liệu từ cổng nối tiếp. Phép đo được thực hiện trong vòng một sợi sinh ra theo cách sau đây: lớp

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(); 

các SerialReader thực hiện Runnable và chỉ lặp vô hạn định, đọc từ cảng và xây dựng dữ liệu thành các gói hữu ích trước khi gửi nó đi vào các ứng dụng khác. Tuy nhiên, tôi đã giảm nó xuống để tính đơn giản sau:

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); 
    } 
    } 
} 

Khi người dùng nhấp vào một nút dừng, đám cháy chức năng sau đó nên về mặt lý thuyết đóng dòng đầu vào và thoát ra khỏi byteChan.read chặn (bộ đệm). Mã này như sau:

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

Tuy nhiên, khi tôi chạy mã này, tôi không bao giờ nhận được một ClosedByInterruptException, NÊN sẽ kích hoạt khi luồng đầu vào đóng. Hơn nữa, các khối thực hiện trên cuộc gọi đến serial.close() - bởi vì luồng đầu vào cơ bản vẫn đang chặn trên cuộc gọi đọc. Tôi đã thử thay thế các cuộc gọi ngắt với byteChan.close(), mà sau đó sẽ gây ra một AsynchronousCloseException, tuy nhiên, tôi nhận được kết quả tương tự.

Bất kỳ trợ giúp nào về những gì tôi thiếu sẽ được đánh giá cao.

Answer 1

Các RXTX SerialInputStream (những gì được trả về bởi các serial.getInputStream() gọi) hỗ trợ một chương trình thời gian chờ mà đã kết thúc giải quyết mọi vấn đề của tôi. Thêm những điều sau trước khi tạo đối tượng SerialReader mới làm cho đọc để không còn chặn vô thời hạn:

serial.enableReceiveTimeout(1000); 

Trong đối tượng SerialReader, tôi đã phải thay đổi một vài điều xung quanh để đọc trực tiếp từ InputStream thay vì tạo các ReadableByteChannel, nhưng bây giờ, tôi có thể dừng lại và khởi động lại trình đọc mà không có vấn đề gì.

Answer 2

Bạn không thể tạo luồng không hỗ trợ I/O gián đoạn thành InterruptibleChannel chỉ đơn giản bằng cách gói nó (và, anyway, ReadableByteChannel không mở rộng InterruptibleChannel).

Bạn phải xem hợp đồng của cơ sở InputStream. SerialPort.getInputStream() nói gì về khả năng gián đoạn kết quả của nó? Nếu nó không nói bất cứ điều gì, bạn nên giả định rằng nó bỏ qua các ngắt.

Đối với bất kỳ I/O nào không hỗ trợ rõ ràng tính gián đoạn, tùy chọn duy nhất thường đóng luồng từ một chuỗi khác. Điều này có thể ngay lập tức tăng IOException (mặc dù nó có thể không phải là một AsynchronousCloseException) trong chuỗi bị chặn khi có cuộc gọi đến luồng.

Tuy nhiên, ngay cả điều này phụ thuộc rất lớn vào việc triển khai InputStream — và hệ điều hành cơ bản cũng có thể là một yếu tố.

---

Note mã nguồn bình luận trên lớp ReadableByteChannelImpl trả về bởi newChannel():

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

Answer 3

tôi đang sử dụng mã bên dưới để tắt rxtx. tôi chạy thử nghiệm mà bắt đầu chúng lên và tắt chúng xuống và dường như làm việc ok.đọc của tôi trông giống như:

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."); 
} 

nhờ

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()); 
    }