Прерывание поток не заканчивается блокировкой вызова на входной поток чтения

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

Однако, когда я запускаю этот код, я никогда не получаю замкнутыйИнтератеррептексию, который должен огонь, когда входной поток закрывается. Кроме того, выполнение блоков на вызов для Serial.close () - потому что базовый входной поток все еще блокирует на вызов чтения. Я пытался заменить вызов прерывания с помощью bytechan.close (), который должен затем вызвать асинхронно -ллюстрацию, однако, я получаю те же результаты.

Любая помощь о том, что мне не хватает, было бы очень оценено.

Answer 1

RXTX SERIALINPUTSTREAM (что возвращается Serial.getInputStream () Call) поддерживает схему тайм-аута, которая заканчивалась от решения всех моих проблем. Добавление следующего перед созданием нового объекта SerialReader приводит к тому, что чтения больше не заблокируются бесконечно:

serial.enableReceiveTimeout(1000);

В объекте SerialReader я должен был изменить несколько вещей, чтобы прочитать непосредственно из ввода вместо того, чтобы создавать ReadableByTeChannel, но теперь я могу остановиться и перезапустить читатель без проблем.

Answer 2

Вы не можете сделать поток, который не поддерживает прерываемые ввода / вывода в InterruptibleChannel просто путем упаковки его (и, во всяком случае, ReadableByteChannel не распространяется InterruptibleChannel).

Вы должны посмотреть на договор в основе базового InputStream. Отказ Что значит SerialPort.getInputStream() Скажите о прерывищении его результата? Если это ничего не скажет, вы должны предположить, что он игнорирует прерывания.

Для любого ввода / вывода, которая не поддерживает прерывание, единственный вариант, как правило, закрывает поток из другого потока. Это может немедленно поднять 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());
    }