Ist es möglich, mit einer Zeitüberschreitung aus einem InputStream zu lesen?

Insbesondere besteht das Problem darin, eine Methode wie die folgende zu schreiben:

int maybeRead(InputStream in, long timeout)

Dabei ist der Rückgabewert derselbe wie in.read (), wenn Daten innerhalb von 'Timeout'-Millisekunden verfügbar sind, andernfalls -2. Bevor die Methode zurückgegeben wird, müssen alle erzeugten Threads beendet werden.

Um Argumente zu vermeiden, wird der Betreff hier java.io.InputStream, wie von Sun dokumentiert (jede Java-Version). Bitte beachten Sie, dass dies nicht so einfach ist, wie es aussieht. Im Folgenden finden Sie einige Fakten, die direkt von der Sun-Dokumentation unterstützt werden.

1. Die Methode in.read () ist möglicherweise nicht unterbrechbar.

2. Das Umschließen des InputStreams in einen Reader oder InterruptibleChannel hilft nicht, da diese Klassen nur Methoden des InputStream aufrufen können. Wenn es möglich wäre, diese Klassen zu verwenden, wäre es möglich, eine Lösung zu schreiben, die dieselbe Logik direkt im InputStream ausführt.

3. Es ist immer akzeptabel, dass in.available () 0 zurückgibt.

4. Die Methode in.close () kann blockieren oder nichts tun.

5. Es gibt keine allgemeine Möglichkeit, einen anderen Thread zu beenden.

Verwenden von inputStream.available ()

Es ist immer akzeptabel, dass System.in.available () 0 zurückgibt.

Ich habe das Gegenteil gefunden - es gibt immer den besten Wert für die Anzahl der verfügbaren Bytes zurück. Javadoc für InputStream.available():

Returns an estimate of the number of bytes that can be read (or skipped over) 
from this input stream without blocking by the next invocation of a method for 
this input stream.

Eine Schätzung ist aufgrund des Timings / der Alterung unvermeidbar. Die Zahl kann eine einmalige Unterschätzung sein, da ständig neue Daten eintreffen. Es "holt" jedoch immer beim nächsten Anruf auf - es sollte alle eingehenden Daten berücksichtigen, abgesehen davon, dass sie gerade zum Zeitpunkt des neuen Anrufs eintreffen. Die dauerhafte Rückgabe von 0, wenn Daten vorhanden sind, schlägt die obige Bedingung fehl.

Erste Einschränkung: Konkrete Unterklassen von InputStream sind für die Verfügbarkeit verantwortlich ()

InputStreamist eine abstrakte Klasse. Es hat keine Datenquelle. Es ist bedeutungslos, Daten zur Verfügung zu haben. Daher heißt es in javadoc available()auch:

The available method for class InputStream always returns 0.

This method should be overridden by subclasses.

Tatsächlich überschreiben die konkreten Eingabestreamklassen die verfügbaren () und liefern aussagekräftige Werte, keine konstanten Nullen.

Zweite Einschränkung: Stellen Sie sicher, dass Sie Wagenrücklauf verwenden, wenn Sie Eingaben in Windows eingeben.

Bei Verwendung System.inerhält Ihr Programm nur Eingaben, wenn Ihre Befehlsshell sie übergibt. Wenn Sie Dateiumleitungen / Pipes verwenden (z. B. somefile> java myJavaApp oder somecommand | java myJavaApp), werden Eingabedaten normalerweise sofort übergeben. Wenn Sie die Eingabe jedoch manuell eingeben, kann sich die Datenübergabe verzögern. Beispiel: Bei der Windows-Shell cmd.exe werden die Daten in der Shell cmd.exe gepuffert. Daten werden erst nach Wagenrücklauf (control-m oder <enter>) an das ausführende Java-Programm übergeben . Dies ist eine Einschränkung der Ausführungsumgebung. Natürlich gibt InputStream.available () 0 zurück, solange die Shell die Daten puffert - das ist korrektes Verhalten. Zu diesem Zeitpunkt sind keine Daten verfügbar. Sobald die Daten in der Shell verfügbar sind, gibt die Methode einen Wert> 0 zurück. Hinweis: Cygwin verwendet cmd.

Einfachste Lösung (keine Blockierung, daher keine Zeitüberschreitung erforderlich)

Verwenden Sie einfach dies:

    byte[] inputData = new byte[1024];
    int result = is.read(inputData, 0, is.available());  
    // result will indicate number of bytes read; -1 for EOF with no data read.

Oder gleichwertig,

    BufferedReader br = new BufferedReader(new InputStreamReader(System.in, Charset.forName("ISO-8859-1")),1024);
    // ...
         // inside some iteration / processing logic:
         if (br.ready()) {
             int readCount = br.read(inputData, bufferOffset, inputData.length-bufferOffset);
         }

Reichhaltigere Lösung (füllt den Puffer innerhalb des Zeitlimits maximal aus)

Erklären Sie dies:

public static int readInputStreamWithTimeout(InputStream is, byte[] b, int timeoutMillis)
     throws IOException  {
     int bufferOffset = 0;
     long maxTimeMillis = System.currentTimeMillis() + timeoutMillis;
     while (System.currentTimeMillis() < maxTimeMillis && bufferOffset < b.length) {
         int readLength = java.lang.Math.min(is.available(),b.length-bufferOffset);
         // can alternatively use bufferedReader, guarded by isReady():
         int readResult = is.read(b, bufferOffset, readLength);
         if (readResult == -1) break;
         bufferOffset += readResult;
     }
     return bufferOffset;
 }

Dann benutze dies:

    byte[] inputData = new byte[1024];
    int readCount = readInputStreamWithTimeout(System.in, inputData, 6000);  // 6 second timeout
    // readCount will indicate number of bytes read; -1 for EOF with no data read.

Angenommen, Ihr Stream wird nicht von einem Socket unterstützt (Sie können ihn also nicht verwenden Socket.setSoTimeout()). Ich denke, die Standardmethode zur Lösung dieser Art von Problem ist die Verwendung einer Zukunft.

Angenommen, ich habe den folgenden Executor und Streams:

    ExecutorService executor = Executors.newFixedThreadPool(2);
    final PipedOutputStream outputStream = new PipedOutputStream();
    final PipedInputStream inputStream = new PipedInputStream(outputStream);

Ich habe einen Writer, der einige Daten schreibt und dann 5 Sekunden wartet, bevor er die letzten Daten schreibt und den Stream schließt:

    Runnable writeTask = new Runnable() {
        @Override
        public void run() {
            try {
                outputStream.write(1);
                outputStream.write(2);
                Thread.sleep(5000);
                outputStream.write(3);
                outputStream.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    };
    executor.submit(writeTask);

Die normale Art, dies zu lesen, ist wie folgt. Der Lesevorgang wird auf unbestimmte Zeit für Daten blockiert und ist daher in 5 Sekunden abgeschlossen:

    long start = currentTimeMillis();
    int readByte = 1;
    // Read data without timeout
    while (readByte >= 0) {
        readByte = inputStream.read();
        if (readByte >= 0)
            System.out.println("Read: " + readByte);
    }
    System.out.println("Complete in " + (currentTimeMillis() - start) + "ms");

welche Ausgänge:

Read: 1
Read: 2
Read: 3
Complete in 5001ms

Wenn es ein grundlegenderes Problem gäbe, wie wenn der Autor nicht antwortet, würde der Leser für immer blockieren. Wenn ich den Lesevorgang in einer Zukunft abschließe, kann ich das Zeitlimit wie folgt steuern:

    int readByte = 1;
    // Read data with timeout
    Callable<Integer> readTask = new Callable<Integer>() {
        @Override
        public Integer call() throws Exception {
            return inputStream.read();
        }
    };
    while (readByte >= 0) {
        Future<Integer> future = executor.submit(readTask);
        readByte = future.get(1000, TimeUnit.MILLISECONDS);
        if (readByte >= 0)
            System.out.println("Read: " + readByte);
    }

welche Ausgänge:

Read: 1
Read: 2
Exception in thread "main" java.util.concurrent.TimeoutException
    at java.util.concurrent.FutureTask$Sync.innerGet(FutureTask.java:228)
    at java.util.concurrent.FutureTask.get(FutureTask.java:91)
    at test.InputStreamWithTimeoutTest.main(InputStreamWithTimeoutTest.java:74)

Ich kann die TimeoutException abfangen und jede gewünschte Bereinigung durchführen.

Wenn Ihr InputStream von einem Socket unterstützt wird, können Sie mit setSoTimeout ein Socket-Timeout (in Millisekunden) festlegen . Wenn der Aufruf von read () nicht innerhalb des angegebenen Zeitlimits entsperrt wird, wird eine SocketTimeoutException ausgelöst.

Stellen Sie einfach sicher, dass Sie setSoTimeout am Socket aufrufen, bevor Sie den Aufruf read () ausführen.

Ich würde die Problemstellung in Frage stellen, anstatt sie nur blind zu akzeptieren. Sie benötigen nur Zeitüberschreitungen von der Konsole oder über das Netzwerk. Wenn Sie letztere haben Socket.setSoTimeout()und HttpURLConnection.setReadTimeout()beide genau das tun, was erforderlich ist, solange Sie sie beim Erstellen / Erwerb korrekt eingerichtet haben. Lassen Sie es später in der Anwendung an einem beliebigen Punkt, wenn Sie nur den InputStream haben. Das schlechte Design führt zu einer sehr umständlichen Implementierung.

Ich habe die Klassen aus dem Java NIO-Paket nicht verwendet, aber es scheint, dass sie hier hilfreich sein könnten. Insbesondere java.nio.channels.Channels und java.nio.channels.InterruptibleChannel .

Hier ist eine Möglichkeit, einen NIO FileChannel von System.in abzurufen und die Verfügbarkeit von Daten mithilfe eines Timeouts zu überprüfen. Dies ist ein Sonderfall des in der Frage beschriebenen Problems. Führen Sie es an der Konsole aus, geben Sie keine Eingabe ein und warten Sie auf die Ergebnisse. Es wurde erfolgreich unter Java 6 unter Windows und Linux getestet.

import java.io.FileInputStream;
import java.io.FilterInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.lang.reflect.Field;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedByInterruptException;

public class Main {

    static final ByteBuffer buf = ByteBuffer.allocate(4096);

    public static void main(String[] args) {

        long timeout = 1000 * 5;

        try {
            InputStream in = extract(System.in);
            if (! (in instanceof FileInputStream))
                throw new RuntimeException(
                        "Could not extract a FileInputStream from STDIN.");

            try {
                int ret = maybeAvailable((FileInputStream)in, timeout);
                System.out.println(
                        Integer.toString(ret) + " bytes were read.");

            } finally {
                in.close();
            }

        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }

    /* unravels all layers of FilterInputStream wrappers to get to the
     * core InputStream
     */
    public static InputStream extract(InputStream in)
            throws NoSuchFieldException, IllegalAccessException {

        Field f = FilterInputStream.class.getDeclaredField("in");
        f.setAccessible(true);

        while( in instanceof FilterInputStream )
            in = (InputStream)f.get((FilterInputStream)in);

        return in;
    }

    /* Returns the number of bytes which could be read from the stream,
     * timing out after the specified number of milliseconds.
     * Returns 0 on timeout (because no bytes could be read)
     * and -1 for end of stream.
     */
    public static int maybeAvailable(final FileInputStream in, long timeout)
            throws IOException, InterruptedException {

        final int[] dataReady = {0};
        final IOException[] maybeException = {null};
        final Thread reader = new Thread() {
            public void run() {                
                try {
                    dataReady[0] = in.getChannel().read(buf);
                } catch (ClosedByInterruptException e) {
                    System.err.println("Reader interrupted.");
                } catch (IOException e) {
                    maybeException[0] = e;
                }
            }
        };

        Thread interruptor = new Thread() {
            public void run() {
                reader.interrupt();
            }
        };

        reader.start();
        for(;;) {

            reader.join(timeout);
            if (!reader.isAlive())
                break;

            interruptor.start();
            interruptor.join(1000);
            reader.join(1000);
            if (!reader.isAlive())
                break;

            System.err.println("We're hung");
            System.exit(1);
        }

        if ( maybeException[0] != null )
            throw maybeException[0];

        return dataReady[0];
    }
}

Interessanterweise funktioniert das Timeout beim Ausführen des Programms in NetBeans 6.5 und nicht an der Konsole überhaupt nicht, und der Aufruf von System.exit () ist tatsächlich erforderlich, um die Zombiethreads zu beenden. Was passiert ist, dass der Interruptor-Thread (!) Beim Aufruf von reader.interrupt () blockiert. Ein anderes Testprogramm (hier nicht gezeigt) versucht zusätzlich, den Kanal zu schließen, aber das funktioniert auch nicht.

Wie gesagt, NIO ist die beste (und richtige) Lösung. Wenn Sie jedoch wirklich mit einem InputStream nicht weiterkommen, können Sie dies auch tun

1. Erstellen Sie einen Thread, dessen exklusive Aufgabe darin besteht, aus dem InputStream zu lesen und das Ergebnis in einen Puffer zu legen, der ohne Blockierung aus Ihrem ursprünglichen Thread gelesen werden kann. Dies sollte gut funktionieren, wenn Sie immer nur eine Instanz des Streams haben. Andernfalls können Sie den Thread möglicherweise mit den veralteten Methoden in der Thread-Klasse beenden. Dies kann jedoch zu Ressourcenlecks führen.

2. Verlassen Sie sich auf isAvailable, um Daten anzuzeigen, die ohne Blockierung gelesen werden können. In einigen Fällen (z. B. bei Sockets) kann es jedoch zu einem potenziell blockierenden Lesevorgang kommen, damit isAvailable etwas anderes als 0 meldet.

Inspiriert von dieser Antwort habe ich eine etwas objektorientiertere Lösung gefunden.

Dies gilt nur, wenn Sie Zeichen lesen möchten

Sie können BufferedReader überschreiben und Folgendes implementieren:

public class SafeBufferedReader extends BufferedReader{

    private long millisTimeout;

    ( . . . )

    @Override
    public int read(char[] cbuf, int off, int len) throws IOException {
        try {
            waitReady();
        } catch(IllegalThreadStateException e) {
            return 0;
        }
        return super.read(cbuf, off, len);
    }

    protected void waitReady() throws IllegalThreadStateException, IOException {
        if(ready()) return;
        long timeout = System.currentTimeMillis() + millisTimeout;
        while(System.currentTimeMillis() < timeout) {
            if(ready()) return;
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                break; // Should restore flag
            }
        }
        if(ready()) return; // Just in case.
        throw new IllegalThreadStateException("Read timed out");
    }
}

Hier ist ein fast vollständiges Beispiel.

Ich gebe bei einigen Methoden 0 zurück. Sie sollten es auf -2 ändern, um Ihre Anforderungen zu erfüllen, aber ich denke, dass 0 für BufferedReader-Verträge besser geeignet ist. Es ist nichts Falsches passiert, es wurden nur 0 Zeichen gelesen. Die readLine-Methode ist ein schrecklicher Leistungskiller. Sie sollten einen völlig neuen BufferedReader erstellen, wenn Sie readLin tatsächlich verwenden möchten e . Im Moment ist es nicht threadsicher. Wenn jemand eine Operation aufruft, während readLines auf eine Zeile wartet, führt dies zu unerwarteten Ergebnissen

Ich mag es nicht, -2 zurückzugeben, wo ich bin. Ich würde eine Ausnahme auslösen, da einige Leute möglicherweise nur prüfen, ob int <0 ist, um EOS zu berücksichtigen. Wie auch immer, diese Methoden behaupten, dass "nicht blockieren kann". Sie sollten überprüfen, ob diese Aussage tatsächlich wahr ist, und sie einfach nicht überschreiben.

import java.io.BufferedReader;
import java.io.IOException;
import java.io.Reader;
import java.nio.CharBuffer;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;

/**
 * 
 * readLine
 * 
 * @author Dario
 *
 */
public class SafeBufferedReader extends BufferedReader{

    private long millisTimeout;

    private long millisInterval = 100;

    private int lookAheadLine;

    public SafeBufferedReader(Reader in, int sz, long millisTimeout) {
        super(in, sz);
        this.millisTimeout = millisTimeout;
    }

    public SafeBufferedReader(Reader in, long millisTimeout) {
        super(in);
        this.millisTimeout = millisTimeout;
    }



    /**
     * This is probably going to kill readLine performance. You should study BufferedReader and completly override the method.
     * 
     * It should mark the position, then perform its normal operation in a nonblocking way, and if it reaches the timeout then reset position and throw IllegalThreadStateException
     * 
     */
    @Override
    public String readLine() throws IOException {
        try {
            waitReadyLine();
        } catch(IllegalThreadStateException e) {
            //return null; //Null usually means EOS here, so we can't.
            throw e;
        }
        return super.readLine();
    }

    @Override
    public int read() throws IOException {
        try {
            waitReady();
        } catch(IllegalThreadStateException e) {
            return -2; // I'd throw a runtime here, as some people may just be checking if int < 0 to consider EOS
        }
        return super.read();
    }

    @Override
    public int read(char[] cbuf) throws IOException {
        try {
            waitReady();
        } catch(IllegalThreadStateException e) {
            return -2;  // I'd throw a runtime here, as some people may just be checking if int < 0 to consider EOS
        }
        return super.read(cbuf);
    }

    @Override
    public int read(char[] cbuf, int off, int len) throws IOException {
        try {
            waitReady();
        } catch(IllegalThreadStateException e) {
            return 0;
        }
        return super.read(cbuf, off, len);
    }

    @Override
    public int read(CharBuffer target) throws IOException {
        try {
            waitReady();
        } catch(IllegalThreadStateException e) {
            return 0;
        }
        return super.read(target);
    }

    @Override
    public void mark(int readAheadLimit) throws IOException {
        super.mark(readAheadLimit);
    }

    @Override
    public Stream<String> lines() {
        return super.lines();
    }

    @Override
    public void reset() throws IOException {
        super.reset();
    }

    @Override
    public long skip(long n) throws IOException {
        return super.skip(n);
    }

    public long getMillisTimeout() {
        return millisTimeout;
    }

    public void setMillisTimeout(long millisTimeout) {
        this.millisTimeout = millisTimeout;
    }

    public void setTimeout(long timeout, TimeUnit unit) {
        this.millisTimeout = TimeUnit.MILLISECONDS.convert(timeout, unit);
    }

    public long getMillisInterval() {
        return millisInterval;
    }

    public void setMillisInterval(long millisInterval) {
        this.millisInterval = millisInterval;
    }

    public void setInterval(long time, TimeUnit unit) {
        this.millisInterval = TimeUnit.MILLISECONDS.convert(time, unit);
    }

    /**
     * This is actually forcing us to read the buffer twice in order to determine a line is actually ready.
     * 
     * @throws IllegalThreadStateException
     * @throws IOException
     */
    protected void waitReadyLine() throws IllegalThreadStateException, IOException {
        long timeout = System.currentTimeMillis() + millisTimeout;
        waitReady();

        super.mark(lookAheadLine);
        try {
            while(System.currentTimeMillis() < timeout) {
                while(ready()) {
                    int charInt = super.read();
                    if(charInt==-1) return; // EOS reached
                    char character = (char) charInt;
                    if(character == '\n' || character == '\r' ) return;
                }
                try {
                    Thread.sleep(millisInterval);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt(); // Restore flag
                    break;
                }
            }
        } finally {
            super.reset();
        }
        throw new IllegalThreadStateException("readLine timed out");

    }

    protected void waitReady() throws IllegalThreadStateException, IOException {
        if(ready()) return;
        long timeout = System.currentTimeMillis() + millisTimeout;
        while(System.currentTimeMillis() < timeout) {
            if(ready()) return;
            try {
                Thread.sleep(millisInterval);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt(); // Restore flag
                break;
            }
        }
        if(ready()) return; // Just in case.
        throw new IllegalThreadStateException("read timed out");
    }

}