La interrupción de subproceso no finaliza la llamada de bloqueo en la secuencia de entrada lee

Question

Estoy usando RXTX para leer datos de un puerto serie. La lectura se realiza dentro de un subproceso generado de la siguiente manera: Clase

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

la SerialReader implementa Ejecutable y simplemente bucles de forma indefinida, la lectura del puerto y la construcción de los datos en paquetes de útiles antes de enviarlo a otras aplicaciones. Sin embargo, he reducido se reduce a lo siguiente simplicidad:

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

Cuando un usuario hace clic en un botón de parada, los siguientes incendios de funcionalidad que en teoría debería cerrar el flujo de entrada y romper el bloqueo (byteChan.read buffer) llamada. El código es el siguiente:

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

Sin embargo, cuando ejecuta este código, nunca consigo una ClosedByInterruptException, que debe disparar una vez que el flujo de entrada se cierra. Además, la ejecución bloquea la llamada a serial.close() porque la secuencia de entrada subyacente todavía está bloqueando en la llamada de lectura. He intentado reemplazar la llamada de interrupción con byteChan.close(), que luego debería causar una AsynchronousCloseException, sin embargo, obtengo los mismos resultados.

Me agradecería cualquier ayuda sobre lo que me estoy perdiendo.

Answer 1

El RXTX SerialInputStream (lo que es devuelto por la llamada serial.getInputStream()) es compatible con un esquema de tiempo de espera terminó resolviendo todos mis problemas Añadiendo lo siguiente antes de crear el nuevo objeto SerialReader hace que la lee a no bloquear indefinidamente:

serial.enableReceiveTimeout(1000); 

dentro del objeto SerialReader, he tenido que cambiar algunas cosas para leer directamente desde el InputStream en lugar de crear el ReadableByteChannel, pero ahora, puedo detener y reiniciar el lector sin problema.

Answer 2

No puede hacer que una secuencia que no admite E/S interrumpible en un InterruptibleChannel simplemente envolviéndola (y, de todos modos, ReadableByteChannel no se extiende InterruptibleChannel).

Tienes que mirar el contrato de InputStream subyacente. ¿Qué dice SerialPort.getInputStream() sobre la interrupción de su resultado? Si no dice nada, debes asumir que ignora las interrupciones.

Para cualquier E/S que no admita explícitamente la interrupción, la única opción generalmente es cerrar la transmisión desde otra secuencia. Esto puede generar inmediatamente un IOException (aunque podría no ser un AsynchronousCloseException) en el hilo bloqueado en una llamada a la transmisión.

Sin embargo, incluso esto depende en gran medida de la implementación del InputStream — y el sistema operativo subyacente también puede ser un factor.

---

Nota del comentario del código fuente de la clase ReadableByteChannelImpl devuelto por newChannel():

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

Answer 3

Estoy usando el siguiente código para apagar rxtx. Ejecuto pruebas que los ponen en marcha y los cierran y parece que funciona bien.mi lector se ve así:

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

gracias

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